import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { Betriebsmodus, LaufStatus, VersionStatus } from "@/generated/prisma/enums"; import type { Rolle } from "@/generated/prisma/enums"; import { HERZSCHLAG, KANAL, herzschlagMs, maxStroeme, sseEreignis, willStrom } from "@/lib/sse"; import { STROM_GESCHLOSSEN, reaktionAufAenderung, sollZurueckfallen, } from "@/components/laeufe/lauf-aktualisierung"; import { RechteFehler } from "@/server/auth/sitzungspflicht"; import type { Sitzung } from "@/lib/sitzung"; import { aufLaufHoeren, ereignisseSchliessen, stroemeOffen } from "@/server/lauf-ereignisse"; import { datenbankLeeren, datenbankVerfuegbar, testPrisma } from "./helpers/db"; /** * Live-Aktualisierung per SSE statt Polling (#033). * * Vier Zusagen tragen das Ganze: die Datenbank meldet **jede** Aenderung, die im * Zustand steht; der Strom prueft Recht und Mandant, bevor das erste Ereignis * fliesst; der Herzschlag haelt die Verbindung offen; und das Polling bleibt ein * echter Rueckfall. * * Die Pruefung „zwei Sitzungen sehen einander in unter einer Sekunde“ steht in * `tests/e2e/live.spec.ts` — hier wird gezeigt, dass die Teile einzeln halten. */ /* Der Handler wird echt gefahren; nur die Sitzung wird gestellt. */ vi.mock("@/server/auth/sitzungspflicht", async (original) => { const echt = await original(); return { ...echt, rechtErforderlich: vi.fn() }; }); const { rechtErforderlich } = await import("@/server/auth/sitzungspflicht"); const { GET } = await import("@/app/api/laeufe/[id]/zustand/route"); describe("Format und Grenzen des Stroms", () => { it("schreibt ein Ereignis im SSE-Format", () => { expect(sseEreignis("zustand", '{"a":1}')).toBe('event: zustand\ndata: {"a":1}\n\n'); }); it("gibt jeder Zeile ihr eigenes data-Feld", () => { /* * Die Falle des Formats: ein Zeilenumbruch beendet das Feld. Ohne diese * Aufteilung zerfiele die Nachricht, und der Client bekäme Schrott. */ expect(sseEreignis("zustand", "erste\nzweite")).toBe("event: zustand\ndata: erste\ndata: zweite\n\n"); }); it("erkennt den Wunsch nach einem Strom am Accept-Kopf", () => { expect(willStrom("text/event-stream")).toBe(true); // So schickt es der Browser wirklich. expect(willStrom("text/event-stream, */*;q=0.1")).toBe(true); expect(willStrom("text/event-stream;charset=utf-8")).toBe(true); expect(willStrom("application/json")).toBe(false); expect(willStrom("*/*")).toBe(false); expect(willStrom(null)).toBe(false); }); it("nimmt Herzschlag und Obergrenze aus der Umgebung, mit Vorgabe", () => { expect(herzschlagMs()).toBe(25_000); expect(maxStroeme()).toBe(200); process.env["SSE_HERZSCHLAG_MS"] = "50"; process.env["SSE_MAX_STROEME"] = "3"; expect(herzschlagMs()).toBe(50); expect(maxStroeme()).toBe(3); // Unsinn fällt auf die Vorgabe zurück, statt den Strom lahmzulegen. process.env["SSE_HERZSCHLAG_MS"] = "0"; process.env["SSE_MAX_STROEME"] = "spaeter"; expect(herzschlagMs()).toBe(25_000); expect(maxStroeme()).toBe(200); delete process.env["SSE_HERZSCHLAG_MS"]; delete process.env["SSE_MAX_STROEME"]; }); it("fällt nur bei endgültig geschlossenem Strom auf Polling zurück", () => { // `EventSource` verbindet sich bei CONNECTING selbst wieder – aufzugeben // wäre verfrüht. expect(sollZurueckfallen(0)).toBe(false); expect(sollZurueckfallen(1)).toBe(false); expect(sollZurueckfallen(STROM_GESCHLOSSEN)).toBe(true); }); it("fragt statt neu zu laden, wenn ein Passwort auf der Seite steht", () => { expect(reaktionAufAenderung(true)).toBe("fragen"); expect(reaktionAufAenderung(false)).toBe("neu-laden"); }); }); async function aufbau() { const kunde = await testPrisma.kunde.create({ data: { name: "Eigene Organisation", kuerzel: "EIGEN", istEigeneOrganisation: true }, }); await testPrisma.einstellung.create({ data: { id: "global", modus: Betriebsmodus.INTERN } }); const admin = await testPrisma.benutzer.create({ data: { email: "admin@firma.example", name: "Admin", rolle: "ADMIN" }, }); const vorlage = await testPrisma.vorlage.create({ data: { name: "PC Neuinstallation" } }); const version = await testPrisma.vorlagenVersion.create({ data: { vorlageId: vorlage.id, version: 1, status: VersionStatus.ENTWURF }, }); const abschnitt = await testPrisma.vorlagenAbschnitt.create({ data: { versionId: version.id, titel: "Vorbereitung", position: 0 }, }); const schritt = await testPrisma.vorlagenSchritt.create({ data: { abschnittId: abschnitt.id, titel: "Rechner auspacken", position: 0, pflicht: true }, }); await testPrisma.vorlagenVersion.update({ where: { id: version.id }, data: { status: VersionStatus.VEROEFFENTLICHT, veroeffentlichtAm: new Date() }, }); const lauf = await testPrisma.lauf.create({ data: { vorlagenVersionId: version.id, titel: "Ein Lauf", kundeId: kunde.id }, }); const ergebnis = await testPrisma.schrittergebnis.create({ data: { laufId: lauf.id, vorlagenSchrittId: schritt.id }, }); return { kunde, admin: { id: admin.id, rolle: admin.rolle }, laufId: lauf.id, schrittId: schritt.id, ergebnisId: ergebnis.id, }; } /** Wartet, bis die Bedingung gilt – oder bricht mit einer lesbaren Meldung ab. */ async function warteAuf(bedingung: () => boolean, was: string, grenzeMs = 5_000): Promise { const ende = Date.now() + grenzeMs; while (Date.now() < ende) { if (bedingung()) return; await new Promise((fertig) => setTimeout(fertig, 10)); } throw new Error(`${was} ist innerhalb von ${grenzeMs} ms nicht eingetreten.`); } describe.runIf(datenbankVerfuegbar)("Lauf-Ereignisse aus der Datenbank", () => { beforeEach(async () => { await datenbankLeeren(); }); afterEach(async () => { await ereignisseSchliessen(); }); afterAll(async () => { await ereignisseSchliessen(); }); it("meldet den Kanalnamen, den die Migration benutzt", async () => { // Liefe der Name auseinander, hörte niemand zu – und nichts wäre rot. const zeilen = await testPrisma.$queryRaw<{ quelle: string }[]>` SELECT prosrc AS quelle FROM pg_proc WHERE proname = 'laufzettel_lauf_geaendert' `; expect(zeilen).toHaveLength(1); expect(zeilen[0]?.quelle).toContain(`pg_notify('${KANAL}'`); }); it("weckt bei einem abgehakten Schritt", async () => { const { laufId, ergebnisId, admin } = await aufbau(); let geweckt = 0; const abmelden = await aufLaufHoeren(laufId, () => { geweckt += 1; }); await testPrisma.schrittergebnis.update({ where: { id: ergebnisId }, data: { zustand: "ERLEDIGT", aktualisiertVonId: admin.id }, }); await warteAuf(() => geweckt > 0, "Die Meldung zum Schrittergebnis"); abmelden(); }); it("weckt auch bei Kommentar, Anhang und Kopfdaten", async () => { /* * **Der eigentliche Rückschritt-Test.** `laufZustand` trägt die Zahl der * Kommentare und Anhänge; das Polling verglich alle zehn Sekunden den * **ganzen** Stand und sah sie deshalb. Ein Trigger nur auf * `Schrittergebnis` wäre gegenüber dem Polling schlechter — ein Kommentar * des Kollegen käme nie an. */ const { laufId, ergebnisId, kunde, admin } = await aufbau(); let geweckt = 0; const abmelden = await aufLaufHoeren(laufId, () => { geweckt += 1; }); const kommentar = await testPrisma.kommentar.create({ data: { laufId, rumpf: "Netzteil fehlte", benutzerId: admin.id }, }); await warteAuf(() => geweckt >= 1, "Die Meldung zum Kommentar"); const vorher = geweckt; const anhang = await testPrisma.anhang.create({ data: { schrittergebnisId: ergebnisId, laufId, kundeId: kunde.id, dateiname: "typenschild.png", medientyp: "image/png", groesse: 10, pruefsumme: "b".repeat(64), }, }); await warteAuf(() => geweckt > vorher, "Die Meldung zum Anhang"); const vorTitel = geweckt; await testPrisma.lauf.update({ where: { id: laufId }, data: { titel: "Neuer Titel" } }); await warteAuf(() => geweckt > vorTitel, "Die Meldung zu den Kopfdaten"); // Und beim Entfernen genauso – sonst bliebe ein gelöschter Anhang stehen. const vorLoeschen = geweckt; await testPrisma.anhang.delete({ where: { id: anhang.id } }); await warteAuf(() => geweckt > vorLoeschen, "Die Meldung zum entfernten Anhang"); const vorKommentarWeg = geweckt; await testPrisma.kommentar.delete({ where: { id: kommentar.id } }); await warteAuf(() => geweckt > vorKommentarWeg, "Die Meldung zum entfernten Kommentar"); abmelden(); }); it("weckt nur die Zuhörer des betroffenen Laufs", async () => { const { laufId, ergebnisId, kunde, admin } = await aufbau(); const version = await testPrisma.vorlagenVersion.findFirstOrThrow(); const anderer = await testPrisma.lauf.create({ data: { vorlagenVersionId: version.id, titel: "Ein anderer Lauf", kundeId: kunde.id }, }); let hier = 0; let dort = 0; const ab1 = await aufLaufHoeren(laufId, () => { hier += 1; }); const ab2 = await aufLaufHoeren(anderer.id, () => { dort += 1; }); await testPrisma.schrittergebnis.update({ where: { id: ergebnisId }, data: { zustand: "ERLEDIGT", aktualisiertVonId: admin.id }, }); await warteAuf(() => hier > 0, "Die Meldung am eigenen Lauf"); // Kurz Luft lassen: eine falsche Zustellung käme in derselben Größenordnung. await new Promise((fertig) => setTimeout(fertig, 200)); expect(dort).toBe(0); ab1(); ab2(); }); it("zählt die offenen Ströme und gibt sie beim Abmelden frei", async () => { const { laufId } = await aufbau(); expect(stroemeOffen()).toBe(0); const ab1 = await aufLaufHoeren(laufId, () => undefined); const ab2 = await aufLaufHoeren(laufId, () => undefined); // Zwei Techniker an einem Lauf sind zwei Verbindungen, nicht eine. expect(stroemeOffen()).toBe(2); ab1(); expect(stroemeOffen()).toBe(1); // Zweimal abmelden darf nicht doppelt zählen. ab1(); expect(stroemeOffen()).toBe(1); ab2(); expect(stroemeOffen()).toBe(0); }); }); /** Liest aus einem Strom, bis `bis` im Gelesenen steht. */ async function stromLesen(antwort: Response, bis: (text: string) => boolean): Promise { const leser = antwort.body?.getReader(); if (!leser) throw new Error("Die Antwort hat keinen Rumpf."); const entschluessler = new TextDecoder(); let gelesen = ""; const ende = Date.now() + 5_000; try { while (Date.now() < ende) { const { value, done } = await leser.read(); if (value) gelesen += entschluessler.decode(value, { stream: true }); if (bis(gelesen)) return gelesen; if (done) return gelesen; } } finally { await leser.cancel().catch(() => undefined); } return gelesen; } describe.runIf(datenbankVerfuegbar)("Der Route Handler", () => { beforeEach(async () => { await datenbankLeeren(); vi.mocked(rechtErforderlich).mockReset(); }); afterEach(async () => { await ereignisseSchliessen(); delete process.env["SSE_HERZSCHLAG_MS"]; delete process.env["SSE_MAX_STROEME"]; }); /** * Die Sitzung wird mit ihrer **echten Form** gestellt, nicht als `{}`. * * Vorher stand hier ein leeres Objekt, und genau daran ist ein Befund * vorbeigekommen: der Strom prueft die Sitzung seit #029 beim Herzschlag nach, * und mit einer Attrappe ohne `benutzerId` liesse sich das nicht zeigen. Eine * Attrappe, die nicht die Form der Wirklichkeit hat, prueft etwas anderes als * die Wirklichkeit — dieselbe Lehre wie beim Graph-Doppelgaenger in #028. */ function sitzungFuer(benutzerId: string, ausgestelltAm = Math.floor(Date.now() / 1000)): Sitzung { return { benutzerId, email: "wer@firma.example", name: "Wer", herkunft: "passwort", ausgestelltAm, }; } function angemeldetAls(benutzer: { id: string; rolle: Rolle }): void { vi.mocked(rechtErforderlich).mockResolvedValue({ benutzer: { id: benutzer.id, rolle: benutzer.rolle }, sitzung: sitzungFuer(benutzer.id), } as unknown as Awaited>); } function anfrage(accept: string | null): Request { return new Request("http://127.0.0.1/api/laeufe/x/zustand", { headers: accept ? { accept } : {}, }); } it("weist ohne Anmeldung ab, bevor das erste Ereignis fließt", async () => { const { laufId } = await aufbau(); vi.mocked(rechtErforderlich).mockRejectedValue(new RechteFehler("Nicht angemeldet.")); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); expect(antwort.status).toBe(401); // Kein Strom: wer kein Recht hat, bekommt keine offene Leitung. expect(antwort.headers.get("content-type")).toContain("application/json"); expect(stroemeOffen()).toBe(0); }); it("liefert für einen Lauf eines fremden Kunden keinen Strom", async () => { const { kunde } = await aufbau(); /* * Die Mandantensicht greift im Modus DIENSTLEISTUNG und fuer eine Rolle * unterhalb von REDAKTEUR – im Modus INTERN sieht jeder alles, und das ist * so gewollt. Der Strom darf diese Grenze nicht umgehen, also wird sie hier * hergestellt statt behauptet. */ await testPrisma.einstellung.update({ where: { id: "global" }, data: { modus: Betriebsmodus.DIENSTLEISTUNG }, }); const vorher = process.env["OIDC_CUSTOMER_GROUP_PREFIX"]; process.env["OIDC_CUSTOMER_GROUP_PREFIX"] = "kunde-"; try { const fremd = await testPrisma.kunde.create({ data: { name: "Fremd GmbH", kuerzel: "FREMD" }, }); const version = await testPrisma.vorlagenVersion.findFirstOrThrow(); const fremderLauf = await testPrisma.lauf.create({ data: { vorlagenVersionId: version.id, titel: "Fremder Lauf", kundeId: fremd.id }, }); // Ein Leser, der nur den eigenen Kunden sieht. const leser = await testPrisma.benutzer.create({ data: { email: "leser@firma.example", name: "Leser", rolle: "LESER" }, }); await testPrisma.benutzerKunde.create({ data: { benutzerId: leser.id, kundeId: kunde.id }, }); vi.mocked(rechtErforderlich).mockResolvedValue({ benutzer: { id: leser.id, rolle: "LESER" }, sitzung: sitzungFuer(leser.id), } as unknown as Awaited>); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: fremderLauf.id }), }); expect(antwort.status).toBe(404); expect(stroemeOffen()).toBe(0); } finally { if (vorher === undefined) delete process.env["OIDC_CUSTOMER_GROUP_PREFIX"]; else process.env["OIDC_CUSTOMER_GROUP_PREFIX"] = vorher; } }); it("antwortet ohne Accept-Kopf wie bisher mit JSON", async () => { const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const antwort = await GET(anfrage(null), { params: Promise.resolve({ id: laufId }) }); expect(antwort.status).toBe(200); expect(antwort.headers.get("content-type")).toContain("application/json"); expect(antwort.headers.get("cache-control")).toBe("no-store"); const stand = (await antwort.json()) as Record; expect(stand["status"]).toBe(LaufStatus.OFFEN); expect(stand["anzahlPflichtOffen"]).toBe(1); // Seit #033 im Stand, damit ein neues Foto überhaupt eine Änderung ist. expect(stand["anzahlAnhaenge"]).toBe(0); expect(stand["anzahlKommentare"]).toBe(0); // Niemals Befehle: ein generiertes Passwort entsteht bei jeder Auflösung neu. expect(JSON.stringify(stand)).not.toContain("befehl"); }); it("schickt im Strom sofort den Stand und danach jede Änderung", async () => { const { laufId, ergebnisId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); expect(antwort.status).toBe(200); expect(antwort.headers.get("content-type")).toContain("text/event-stream"); // Ohne diese Kopfzeile sammelt nginx den Strom und gibt ihn erst am Ende heraus. expect(antwort.headers.get("x-accel-buffering")).toBe("no"); const rumpf = antwort.body; if (!rumpf) throw new Error("Kein Rumpf."); const leser = rumpf.getReader(); const entschluessler = new TextDecoder(); async function lesenBis(bedingung: (text: string) => boolean): Promise { let gelesen = ""; const ende = Date.now() + 5_000; while (Date.now() < ende) { const { value, done } = await leser.read(); if (value) gelesen += entschluessler.decode(value, { stream: true }); if (bedingung(gelesen)) return gelesen; if (done) break; } throw new Error(`Nichts Passendes im Strom. Gelesen: ${JSON.stringify(gelesen)}`); } const erstes = await lesenBis((text) => text.includes("event: zustand")); expect(erstes).toContain('"anzahlPflichtOffen":1'); // Jetzt die Änderung – sie muss von selbst durchkommen. await testPrisma.schrittergebnis.update({ where: { id: ergebnisId }, data: { zustand: "ERLEDIGT", aktualisiertVonId: admin.id }, }); const zweites = await lesenBis((text) => text.includes('"anzahlPflichtOffen":0')); expect(zweites).toContain("ERLEDIGT"); await leser.cancel(); }); it("hält die Verbindung mit einem Herzschlag offen", async () => { // 60 s echt zu warten wäre ein Test, den niemand mehr fährt. process.env["SSE_HERZSCHLAG_MS"] = "40"; const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); const gelesen = await stromLesen(antwort, (text) => text.split(HERZSCHLAG).length - 1 >= 3); // Drei Herzschläge ohne jede Änderung am Lauf. expect(gelesen.split(HERZSCHLAG).length - 1).toBeGreaterThanOrEqual(3); }); it("beendet den Strom, wenn das Konto deaktiviert wird", async () => { /* * **Ein offener Strom hat das Deaktivieren überlebt** (Reviewbefund zu * #029). Das Recht wurde genau einmal geprüft, beim Aufbau, und das Paar aus * Kennung und Rolle für die Lebensdauer des Stroms festgehalten. Ein * Administrator deaktivierte ein Konto — und eine offen gelassene Lasche auf * dem Schreibtisch des Betroffenen bekam weiter live Status, Schrittzustände * und die Zahl der Kommentare, bis jemand sie schloss. * * `docs/architektur.md` verspricht „ein deaktivierter Benutzer wird auch mit * gültigem Cookie abgewiesen“. Für einen offenen Strom galt das nicht. */ process.env["SSE_HERZSCHLAG_MS"] = "40"; const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); await testPrisma.benutzer.update({ where: { id: admin.id }, data: { aktiv: false } }); const gelesen = await stromLesen(antwort, (text) => text.includes("nicht-angemeldet")); expect(gelesen).toContain("nicht-angemeldet"); // Und der Strom ist auch wirklich freigegeben, nicht nur stillgelegt. await warteAuf(() => stroemeOffen() === 0, "Der Strom nach dem Deaktivieren"); }); it("beendet den Strom, wenn die Sitzungen widerrufen werden", async () => { /* * Derselbe Weg, anderer Anlass: ein Administrator beendet die Sitzungen * eines Benutzers (`sitzungenUngueltigAb`), ohne das Konto zu deaktivieren — * etwa nach einem verlorenen Notebook. Dann muss der Strom genauso enden. */ process.env["SSE_HERZSCHLAG_MS"] = "40"; const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); await testPrisma.benutzer.update({ where: { id: admin.id }, // Eine Minute in der Zukunft: die Sitzung dieses Tests ist „von vorher“. data: { sitzungenUngueltigAb: new Date(Date.now() + 60_000) }, }); expect(await stromLesen(antwort, (text) => text.includes("nicht-angemeldet"))).toContain( "nicht-angemeldet", ); }); it("hält den Strom, solange das Konto gilt", async () => { /* * Die Gegenrichtung, und sie ist nötig: eine Nachprüfung, die **immer** * beendet, sähe im Test oben genauso aus. Hier bleibt der Strom über * mehrere Herzschläge offen, ohne dass sich am Konto etwas ändert. */ process.env["SSE_HERZSCHLAG_MS"] = "40"; const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const antwort = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); const gelesen = await stromLesen(antwort, (text) => text.split(HERZSCHLAG).length - 1 >= 3); expect(gelesen).not.toContain("nicht-angemeldet"); expect(gelesen.split(HERZSCHLAG).length - 1).toBeGreaterThanOrEqual(3); }); it("fällt bei erreichter Obergrenze auf eine gewöhnliche Antwort zurück", async () => { process.env["SSE_MAX_STROEME"] = "1"; const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const erster = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); expect(erster.headers.get("content-type")).toContain("text/event-stream"); await warteAuf(() => stroemeOffen() === 1, "Der erste Strom"); /* * Der zweite bekommt JSON statt eines Stroms. `EventSource` meldet darauf * `error` mit `readyState === CLOSED` und versucht es nicht wieder — der * Client fällt damit von sich aus auf Polling zurück, ohne Sonderfall. */ const zweiter = await GET(anfrage("text/event-stream"), { params: Promise.resolve({ id: laufId }), }); expect(zweiter.status).toBe(200); expect(zweiter.headers.get("content-type")).toContain("application/json"); expect(stroemeOffen()).toBe(1); await erster.body?.cancel(); }); it("gibt den Strom frei, wenn der Browser abbricht", async () => { const { laufId, admin } = await aufbau(); angemeldetAls(admin as { id: string; rolle: "ADMIN" }); const steuerung = new AbortController(); const antwort = await GET( new Request("http://127.0.0.1/api/laeufe/x/zustand", { headers: { accept: "text/event-stream" }, signal: steuerung.signal, }), { params: Promise.resolve({ id: laufId }) }, ); expect(antwort.headers.get("content-type")).toContain("text/event-stream"); await warteAuf(() => stroemeOffen() === 1, "Der Strom"); /* * Ohne Freigabe wüchse das Register mit jeder geschlossenen Lasche weiter, * und die Obergrenze zählte irgendwann Verbindungen, die es nicht gibt. */ steuerung.abort(); await warteAuf(() => stroemeOffen() === 0, "Die Freigabe des Stroms"); }); });