Merge branch 'phase/007-worker-vorgaenge'

2026-10-12 09:21:41 +02:00
parent 63a378df8f 4a23930b54
commit a2bdd9f205
16 changed files with 567 additions and 9 deletions
@@ -0,0 +1,57 @@
+import type { Prisma } from "@/generated/prisma/client";
+import type { VorgangsArt } from "@/generated/prisma/enums";
+import { prisma } from "@/lib/db";
+import { jobSchicken } from "@/server/vorgaenge/warteschlange";
+
+/**
+ * Vorgaenge anlegen und lesen (#007). Ein Vorgang ist die fachliche Klammer um einen Job:
+ * die Oberflaeche zeigt ihn an, der Worker fuehrt ihn aus.
+ */
+export type VorgangAnlegen = {
+  readonly art: VorgangsArt;
+  readonly angefordertVonId: string | null;
+  readonly hostId?: string | null;
+  readonly schluesselId?: string | null;
+  /** Nie Geheimnisse – die Eingabe ist in der Oberflaeche sichtbar. */
+  readonly eingabe?: Prisma.InputJsonValue;
+  readonly geheim?: string;
+};
+
+export async function vorgangEinreihen(daten: VorgangAnlegen): Promise<string> {
+  const vorgang = await prisma.vorgang.create({
+    data: {
+      art: daten.art,
+      angefordertVonId: daten.angefordertVonId,
+      hostId: daten.hostId ?? null,
+      schluesselId: daten.schluesselId ?? null,
+      eingabe: daten.eingabe,
+    },
+  });
+  const jobId = await jobSchicken({
+    vorgangId: vorgang.id,
+    ...(daten.geheim ? { geheim: daten.geheim } : {}),
+  });
+  await prisma.vorgang.update({ where: { id: vorgang.id }, data: { jobId } });
+  return vorgang.id;
+}
+
+/** Was die Oberflaeche ueber einen Vorgang erfahren darf – ohne Job-Interna. */
+export async function vorgangLesen(id: string) {
+  return prisma.vorgang.findUnique({
+    where: { id },
+    select: {
+      id: true,
+      art: true,
+      status: true,
+      hostId: true,
+      schluesselId: true,
+      eingabe: true,
+      plan: true,
+      ergebnis: true,
+      fehler: true,
+      angelegtAm: true,
+      gestartetAm: true,
+      beendetAm: true,
+    },
+  });
+}
@@ -0,0 +1,62 @@
+import { PgBoss } from "pg-boss";
+
+/**
+ * Gemeinsame Anbindung an pg-boss fuer Web und Worker (#007).
+ *
+ * Der Webprozess **schickt** nur; abgearbeitet wird ausschliesslich im Worker. Beide
+ * teilen die Datenbank, pg-boss legt sein Schema `pgboss` daneben an.
+ */
+
+/** Eine Queue je Vorgangsart haelt Wiederholungen und Gleichzeitigkeit getrennt einstellbar. */
+export const WARTESCHLANGE = "sshkv-vorgang";
+export const ZEITPLAN_ABGLEICH = "sshkv-abgleich-zeitplan";
+
+/** Was im Job steht. Absichtlich nur die ID: alles Weitere liegt im `Vorgang`. */
+export type VorgangsJob = {
+  readonly vorgangId: string;
+  /**
+   * Nur fuer den Bootstrap per Passwort (#010): mit SSHKV_APP_SECRET verschluesselt. Der
+   * Worker loescht den Job nach dem Lauf, damit auch das Chiffrat nicht liegen bleibt.
+   */
+  readonly geheim?: string;
+};
+
+let instanz: Promise<PgBoss> | undefined;
+
+function verbindung(): string {
+  const url = process.env["DATABASE_URL"];
+  if (!url) throw new Error("DATABASE_URL ist nicht gesetzt. Siehe .env.example.");
+  return url;
+}
+
+/**
+ * Lazy und einmal je Prozess – wie der Prisma-Client, damit `next build` ohne Datenbank
+ * durchlaeuft und Hot Reload keine Verbindungen stapelt.
+ */
+export function warteschlangeHolen(): Promise<PgBoss> {
+  const zwischenspeicher = globalThis as typeof globalThis & { sshkvBoss?: Promise<PgBoss> };
+  const vorhanden = process.env["NODE_ENV"] !== "production" ? zwischenspeicher.sshkvBoss : instanz;
+  if (vorhanden) return vorhanden;
+
+  const neu = (async () => {
+    const boss = new PgBoss({ connectionString: verbindung(), schema: "pgboss" });
+    boss.on("error", (fehler: Error) => console.error("pg-boss:", fehler.message));
+    await boss.start();
+    await boss.createQueue(WARTESCHLANGE, {
+      // Kein automatisches Wiederholen: ein halb ausgefuehrter Schreibvorgang auf einem Host
+      // soll ein Mensch ansehen, nicht der Worker blind noch einmal starten.
+      retryLimit: 0,
+      // Abgeschlossene Jobs nicht lange aufheben – der Verlauf steht im Vorgang.
+      deleteAfterSeconds: 60 * 60,
+    });
+    return boss;
+  })();
+  if (process.env["NODE_ENV"] !== "production") zwischenspeicher.sshkvBoss = neu;
+  else instanz = neu;
+  return neu;
+}
+
+export async function jobSchicken(job: VorgangsJob): Promise<string | null> {
+  const boss = await warteschlangeHolen();
+  return boss.send(WARTESCHLANGE, job);
+}
@@ -0,0 +1,84 @@
+import type { Prisma, Vorgang } from "@/generated/prisma/client";
+import type { VorgangsArt } from "@/generated/prisma/enums";
+import { prisma } from "@/lib/db";
+
+/**
+ * Fuehrt einen Vorgang aus (#007): Status setzen, Handler rufen, Ergebnis oder Fehler
+ * festhalten. Ohne pg-boss geschrieben, damit es sich mit einer Handler-Tabelle testen
+ * laesst; src/worker/index.ts verdrahtet es mit der Warteschlange.
+ */
+export type HandlerErgebnis = {
+  readonly status?: "ERLEDIGT" | "WARTET_AUF_BESTAETIGUNG";
+  readonly ergebnis?: Prisma.InputJsonValue;
+  readonly plan?: Prisma.InputJsonValue;
+};
+
+export type HandlerKontext = {
+  readonly vorgang: Vorgang;
+  /** Entschluesselbares Geheimnis aus dem Job (nur Bootstrap per Passwort). */
+  readonly geheim?: string;
+};
+
+export type Handler = (kontext: HandlerKontext) => Promise<HandlerErgebnis>;
+export type HandlerTabelle = Partial<Record<VorgangsArt, Handler>>;
+
+/**
+ * Fehler, deren Meldung fuer die Oberflaeche gedacht ist. Alle anderen werden zu einer
+ * allgemeinen Meldung – ein Stacktrace oder eine ssh2-Meldung kann Pfade, Adressen oder
+ * im schlimmsten Fall Schluesselmaterial tragen.
+ */
+export class VorgangsFehler extends Error {
+  constructor(nachricht: string, optionen?: ErrorOptions) {
+    super(nachricht, optionen);
+    this.name = "VorgangsFehler";
+  }
+}
+
+export function fehlerText(fehler: unknown): string {
+  if (fehler instanceof VorgangsFehler) return fehler.message;
+  return "Interner Fehler im Worker. Details stehen im Worker-Log.";
+}
+
+export async function vorgangAusfuehren(
+  vorgangId: string,
+  handler: HandlerTabelle,
+  geheim?: string,
+): Promise<void> {
+  // Nur WARTEND darf starten. Das bedingte Update ist zugleich die Sperre gegen einen
+  // zweiten Worker, der denselben Job bekommt.
+  const gestartet = await prisma.vorgang.updateMany({
+    where: { id: vorgangId, status: "WARTEND" },
+    data: { status: "LAEUFT", gestartetAm: new Date() },
+  });
+  if (gestartet.count === 0) return;
+  const vorgang = await prisma.vorgang.findUniqueOrThrow({ where: { id: vorgangId } });
+
+  const ausfuehrer = handler[vorgang.art];
+  try {
+    if (!ausfuehrer) {
+      throw new VorgangsFehler(`Für Vorgänge der Art ${vorgang.art} gibt es noch keine Ausführung.`);
+    }
+    const ergebnis = await ausfuehrer({ vorgang, ...(geheim ? { geheim } : {}) });
+    await prisma.vorgang.update({
+      where: { id: vorgangId },
+      data: {
+        status: ergebnis.status ?? "ERLEDIGT",
+        ergebnis: ergebnis.ergebnis,
+        plan: ergebnis.plan,
+        beendetAm: ergebnis.status === "WARTET_AUF_BESTAETIGUNG" ? null : new Date(),
+      },
+    });
+  } catch (fehler) {
+    if (!(fehler instanceof VorgangsFehler)) {
+      // Ins Log nur Name und Meldung, nie das ganze Objekt (cause-Ketten, Puffer).
+      console.error(
+        `Vorgang ${vorgangId} (${vorgang.art}) fehlgeschlagen:`,
+        fehler instanceof Error ? `${fehler.name}: ${fehler.message}` : String(fehler),
+      );
+    }
+    await prisma.vorgang.update({
+      where: { id: vorgangId },
+      data: { status: "FEHLGESCHLAGEN", fehler: fehlerText(fehler), beendetAm: new Date() },
+    });
+  }
+}
@@ -0,0 +1,7 @@
+import type { HandlerTabelle } from "@/worker/ausfuehren";
+
+/**
+ * Welche Vorgangsart wer ausfuehrt. Waechst mit den Issues: Host-Key (#009), Bootstrap
+ * (#010), Ausrollen (#016), Abgleich (#021) …
+ */
+export const HANDLER: HandlerTabelle = {};
@@ -0,0 +1,43 @@
+import "dotenv/config";
+import type { Job } from "pg-boss";
+import { VERSION } from "@/lib/version";
+import { WARTESCHLANGE, warteschlangeHolen, type VorgangsJob } from "@/server/vorgaenge/warteschlange";
+import { vorgangAusfuehren } from "@/worker/ausfuehren";
+import { HANDLER } from "@/worker/handler";
+
+/**
+ * Einstieg des Worker-Prozesses (#007). Nur hier liegen Master-Key und Deployment-Key,
+ * und nur von hier aus spricht sshkv SSH mit den Hosts.
+ */
+async function starten(): Promise<void> {
+  const boss = await warteschlangeHolen();
+
+  await boss.work<VorgangsJob>(
+    WARTESCHLANGE,
+    { batchSize: 1, localConcurrency: 4 },
+    async (jobs: Job<VorgangsJob>[]) => {
+      for (const job of jobs) {
+        await vorgangAusfuehren(job.data.vorgangId, HANDLER, job.data.geheim);
+        if (job.data.geheim) {
+          // Das verschluesselte Geheimnis soll nicht bis zur Aufraeumrunde liegen bleiben.
+          await boss.deleteJob(WARTESCHLANGE, job.id).catch(() => undefined);
+        }
+      }
+    },
+  );
+
+  console.log(`sshkv-worker ${VERSION} bereit.`);
+
+  const beenden = async (signal: string) => {
+    console.log(`${signal} empfangen, beende laufende Vorgänge …`);
+    await boss.stop({ graceful: true, timeout: 30_000 });
+    process.exit(0);
+  };
+  process.on("SIGTERM", () => void beenden("SIGTERM"));
+  process.on("SIGINT", () => void beenden("SIGINT"));
+}
+
+starten().catch((fehler: unknown) => {
+  console.error("Worker konnte nicht starten:", fehler instanceof Error ? fehler.message : String(fehler));
+  process.exit(1);
+});