feat(worker): Worker mit pg-boss, Vorgaenge und Docker-Betrieb (#007)
Die Oberflaeche legt einen Vorgang an und schickt nur dessen ID in die Warteschlange; der Worker fuehrt ihn aus und haelt Status, Ergebnis oder Fehler fest. So bleiben lange SSH-Vorgaenge aus dem Webprozess heraus, und der Fortschritt ist ohne Job-Interna sichtbar. Kein automatisches Wiederholen: ein halb ausgefuehrter Schreibvorgang auf einem Host gehoert vor einen Menschen. Fehlertexte aus fremden Bibliotheken erreichen die Oberflaeche nie. Dockerfile und Compose fuer den Betrieb: Master-Key als Secret nur im Worker, Datenbank in einem internen Netz. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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);
|
||||
}
|
||||
Reference in New Issue
Block a user