feat(auftraege): worker, Phasen und Live-Fortschritt per SSE

Aufträge laufen als pg-boss-Jobs im worker. Der Phasen-Runner der Engine kennt weder Proxmox noch Datenbank; eine Senke schreibt Phasen und Log in die DB und meldet per NOTIFY, die app streamt per SSE an Web und CLI (Last-Event-ID, Herzschlag, Polling-Fallback im Browser). VMIDs vergibt eine Advisory-Sperre je Server zusammen mit einer Reservierungstabelle; zehn parallele Reservierungen im Test bekommen zehn verschiedene IDs. Der Proxmox-Client hat einen Trockenmodus für den Dry-Run: Lesendes echt, Schreibendes nur aufgezeichnet. Live-Ansicht im Aufbau der CLI-Ausgabe. Im echten Betrieb aufgefallen: ssh2 und pg sind CommonJS und brauchen im worker Default-Importe. Closes #006 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 10:54:41 +02:00
parent 01d22aac3a
commit 76bc0e4269
38 changed files with 2274 additions and 22 deletions
@@ -0,0 +1,270 @@
+import { afterAll, beforeEach, describe, expect, it } from "vitest";
+import { GET as ereignisseGet } from "@/app/api/v1/auftraege/[id]/ereignisse/route";
+import { GET as auftragGet } from "@/app/api/v1/auftraege/[id]/route";
+import { POST as abbrechenPost } from "@/app/api/v1/auftraege/[id]/abbrechen/route";
+import { verschluesseln } from "@/lib/geheimnisse";
+import { prisma } from "@/lib/db";
+import { phasenAusfuehren } from "@/engine/ablauf/ablauf";
+import { auftragAnlegen } from "@/server/services/auftraege";
+import { vmidReservieren } from "@/server/services/vmid";
+import { ablaufRegistrieren, auftragAusfuehren, verwaisteAuftraegeBeenden } from "@/worker/ausfuehren";
+import { datenbankLeeren, datenbankVerfuegbar, testPrisma } from "./helpers/db";
+import { angemeldet, anfrage, kontext } from "./helpers/anfragen";
+
+async function serverAnlegen() {
+  return await testPrisma.server.create({
+    data: {
+      name: "rz1",
+      titel: "RZ 1",
+      url: "https://127.0.0.1:8006",
+      tokenId: "a@pve!b",
+      tokenGeheimnis: verschluesseln("x"),
+      fingerabdruck: "AA",
+    },
+  });
+}
+
+/** Liest einen SSE-Strom, bis `ende` kommt, und gibt die Ereignisse zurück. */
+async function stromLesen(
+  antwort: Response,
+  nebenbei?: () => Promise<void>,
+): Promise<Array<{ art: string; daten: unknown; id?: string }>> {
+  const leser = antwort.body!.getReader();
+  const dekodierer = new TextDecoder();
+  let puffer = "";
+  const ereignisse: Array<{ art: string; daten: unknown; id?: string }> = [];
+  const parallel = nebenbei?.();
+  for (;;) {
+    const { value, done } = await leser.read();
+    if (done) break;
+    puffer += dekodierer.decode(value, { stream: true });
+    let grenze: number;
+    while ((grenze = puffer.indexOf("\n\n")) >= 0) {
+      const block = puffer.slice(0, grenze);
+      puffer = puffer.slice(grenze + 2);
+      const art = /^event: (.+)$/m.exec(block)?.[1];
+      const daten = /^data: (.+)$/m.exec(block)?.[1];
+      const id = /^id: (.+)$/m.exec(block)?.[1];
+      if (art && daten) ereignisse.push({ art, daten: JSON.parse(daten), id });
+    }
+  }
+  await parallel;
+  return ereignisse;
+}
+
+describe.runIf(datenbankVerfuegbar)("Aufträge, worker und Live-Strom", () => {
+  afterAll(async () => {
+    await prisma.$disconnect();
+  });
+  beforeEach(async () => {
+    await datenbankLeeren();
+  });
+
+  it("VMID-Vergabe: zehn parallele Reservierungen bekommen zehn verschiedene IDs, Belegtes wird übersprungen", async () => {
+    const server = await serverAnlegen();
+    const ids = await Promise.all(
+      Array.from({ length: 10 }, (_, i) =>
+        vmidReservieren({
+          serverId: server.id,
+          auftragId: `a${i}`,
+          belegt: new Set([100, 101, 103]),
+          von: 100,
+        }),
+      ),
+    );
+    expect(new Set(ids).size).toBe(10);
+    expect(ids).not.toContain(100);
+    expect(ids).not.toContain(103);
+    expect(Math.min(...ids)).toBe(102);
+    await expect(
+      vmidReservieren({ serverId: server.id, auftragId: "x", belegt: new Set(), von: 100, wunsch: 102 }),
+    ).rejects.toThrow(/bereits belegt/);
+  });
+
+  it("worker führt einen Auftrag aus; SSE liefert Stand, Log und Ende; Reservierungen werden frei", async () => {
+    const server = await serverAnlegen();
+    const { benutzer, cookie } = await angemeldet("OPERATOR");
+    ablaufRegistrieren("AUFRAEUMEN", async ({ auftrag, senke }) => {
+      await vmidReservieren({
+        serverId: auftrag.serverId,
+        auftragId: auftrag.id,
+        belegt: new Set(),
+        von: 100,
+      });
+      const ergebnis = await phasenAusfuehren(
+        [
+          {
+            schluessel: "eins",
+            titel: "Eins",
+            ausfuehren: async (k) => {
+              await k.log("Zeile 1");
+              return { detail: "fertig" };
+            },
+          },
+          {
+            schluessel: "zwei",
+            titel: "Zwei",
+            ausfuehren: async (k) => {
+              await k.log("Zeile 2");
+              return {};
+            },
+          },
+        ],
+        senke,
+      );
+      return { ...ergebnis, ergebnis: { ip: "10.0.0.5" } };
+    });
+    const auftrag = await auftragAnlegen(
+      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
+      { benutzer, quelle: "WEB" },
+      null,
+      async () => {},
+    );
+    expect(auftrag.kurzId).toMatch(/^c-[2-9a-z]{5}$/);
+
+    const strom = await ereignisseGet(
+      anfrage("GET", `/api/v1/auftraege/${auftrag.kurzId}/ereignisse`, cookie),
+      kontext({ id: auftrag.kurzId }),
+    );
+    expect(strom.headers.get("content-type")).toContain("text/event-stream");
+    const ereignisse = await stromLesen(strom, () => auftragAusfuehren(auftrag.id));
+
+    const staende = ereignisse
+      .filter((e) => e.art === "stand")
+      .map((e) => (e.daten as { status: string }).status);
+    // Je nach Takt sieht der erste Stand den Auftrag wartend oder schon laufend – nie fertig.
+    expect(["WARTEND", "LAEUFT"]).toContain(staende[0]);
+    expect(staende.at(-1)).toBe("ERFOLGREICH");
+    const logs = ereignisse
+      .filter((e) => e.art === "log")
+      .flatMap((e) => (e.daten as Array<{ nachricht: string }>).map((z) => z.nachricht));
+    expect(logs).toEqual(["Zeile 1", "Zeile 2"]);
+    expect(ereignisse.at(-1)?.art).toBe("ende");
+
+    const fertig = await (
+      await auftragGet(
+        anfrage("GET", `/api/v1/auftraege/${auftrag.kurzId}`, cookie),
+        kontext({ id: auftrag.kurzId }),
+      )
+    ).json();
+    expect(
+      fertig.phasen.map((p: { schluessel: string; status: string }) => `${p.schluessel}:${p.status}`),
+    ).toEqual(["eins:ERLEDIGT", "zwei:ERLEDIGT"]);
+    expect(fertig.ergebnis).toEqual({ ip: "10.0.0.5" });
+    expect(await testPrisma.vmidReservierung.count()).toBe(0);
+  });
+
+  it("Wiederverbinden mit Last-Event-ID liefert nur neue Log-Zeilen", async () => {
+    const server = await serverAnlegen();
+    const { benutzer, cookie } = await angemeldet("OPERATOR");
+    ablaufRegistrieren("AUFRAEUMEN", async ({ senke }) =>
+      phasenAusfuehren(
+        [
+          {
+            schluessel: "a",
+            titel: "A",
+            ausfuehren: async (k) => {
+              await k.log("alt");
+              await k.log("neu");
+              return {};
+            },
+          },
+        ],
+        senke,
+      ),
+    );
+    const auftrag = await auftragAnlegen(
+      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
+      { benutzer, quelle: "WEB" },
+      null,
+      async () => {},
+    );
+    await auftragAusfuehren(auftrag.id);
+    const erste = await testPrisma.logeintrag.findFirstOrThrow({
+      where: { auftragId: auftrag.id, nachricht: "alt" },
+    });
+
+    const strom = await ereignisseGet(
+      new Request(`http://localhost/api/v1/auftraege/${auftrag.kurzId}/ereignisse`, {
+        headers: { cookie, "last-event-id": erste.id.toString() },
+      }),
+      kontext({ id: auftrag.kurzId }),
+    );
+    const ereignisse = await stromLesen(strom);
+    const logs = ereignisse
+      .filter((e) => e.art === "log")
+      .flatMap((e) => (e.daten as Array<{ nachricht: string }>).map((z) => z.nachricht));
+    expect(logs).toEqual(["neu"]);
+  });
+
+  it("Abbruch: Operator fordert an, der Ablauf endet vor der nächsten Phase als ABGEBROCHEN", async () => {
+    const server = await serverAnlegen();
+    const { benutzer, cookie } = await angemeldet("OPERATOR");
+    let auftragKurz = "";
+    ablaufRegistrieren("AUFRAEUMEN", async ({ senke, abbruchGewuenscht }) =>
+      phasenAusfuehren(
+        [
+          {
+            schluessel: "a",
+            titel: "A",
+            ausfuehren: async () => {
+              await abbrechenPost(
+                anfrage("POST", `/api/v1/auftraege/${auftragKurz}/abbrechen`, cookie),
+                kontext({ id: auftragKurz }),
+              );
+              return {};
+            },
+          },
+          { schluessel: "b", titel: "B", ausfuehren: async () => ({}) },
+        ],
+        senke,
+        { abbruchGewuenscht },
+      ),
+    );
+    const auftrag = await auftragAnlegen(
+      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
+      { benutzer, quelle: "WEB" },
+      null,
+      async () => {},
+    );
+    auftragKurz = auftrag.kurzId;
+    await auftragAusfuehren(auftrag.id);
+    const fertig = await testPrisma.auftrag.findUniqueOrThrow({
+      where: { id: auftrag.id },
+      include: { phasen: { orderBy: { reihenfolge: "asc" } } },
+    });
+    expect(fertig.status).toBe("ABGEBROCHEN");
+    expect(fertig.phasen.map((p) => p.status)).toEqual(["ERLEDIGT", "FEHLER"]);
+    expect(await testPrisma.protokoll.count({ where: { aktion: "Abbruch angefordert" } })).toBe(1);
+  });
+
+  it("ein doppelt gelieferter Job läuft nur einmal; verwaiste Aufträge werden beim Start beendet", async () => {
+    const server = await serverAnlegen();
+    const { benutzer } = await angemeldet("OPERATOR");
+    let laeufe = 0;
+    ablaufRegistrieren("AUFRAEUMEN", async () => {
+      laeufe += 1;
+      return { ok: true };
+    });
+    const auftrag = await auftragAnlegen(
+      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
+      { benutzer, quelle: "WEB" },
+      null,
+      async () => {},
+    );
+    await Promise.all([auftragAusfuehren(auftrag.id), auftragAusfuehren(auftrag.id)]);
+    expect(laeufe).toBe(1);
+
+    const haengt = await auftragAnlegen(
+      { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} },
+      { benutzer, quelle: "WEB" },
+      null,
+      async () => {},
+    );
+    await testPrisma.auftrag.update({ where: { id: haengt.id }, data: { status: "LAEUFT" } });
+    expect(await verwaisteAuftraegeBeenden()).toBe(1);
+    expect((await testPrisma.auftrag.findUniqueOrThrow({ where: { id: haengt.id } })).status).toBe(
+      "FEHLGESCHLAGEN",
+    );
+  });
+});