Laufzettel/tests/lauf-ereignisse.test.ts

import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { Betriebsmodus, LaufStatus, VersionStatus } 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 { 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<typeof import("@/server/auth/sitzungspflicht")>();
  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<void> {
  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<string> {
  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"];
  });

  function angemeldetAls(benutzer: { id: string; rolle: "ADMIN" }): void {
    vi.mocked(rechtErforderlich).mockResolvedValue({
      benutzer: { id: benutzer.id, rolle: benutzer.rolle },
      sitzung: {},
    } as unknown as Awaited<ReturnType<typeof rechtErforderlich>>);
  }

  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: {},
      } as unknown as Awaited<ReturnType<typeof rechtErforderlich>>);

      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<string, unknown>;
    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<string> {
      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("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");
  });
});