108 lines
4.5 KiB
TypeScript
108 lines
4.5 KiB
TypeScript
import "dotenv/config";
|
||
import type { Job } from "pg-boss";
|
||
import { VERSION } from "@/lib/version";
|
||
import { alleHostsAbgleichen } from "@/server/services/abgleich";
|
||
import {
|
||
WARTESCHLANGE,
|
||
ZEITPLAN_ABGLEICH,
|
||
ZEITPLAN_MELDUNG,
|
||
ZEITPLAN_MELDUNG_TAKT,
|
||
warteschlangeHolen,
|
||
type VorgangsJob,
|
||
} from "@/server/vorgaenge/warteschlange";
|
||
import { vorgangAusfuehren } from "@/worker/ausfuehren";
|
||
import { abholungenAufraeumen } from "@/worker/ablaeufe/privatekey";
|
||
import { taktVersenden, zusammenfassungVersenden } from "@/worker/benachrichtigung/versand";
|
||
import { deploymentKeySicherstellen, masterKeyPruefen } from "@/worker/geheimnisse";
|
||
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> {
|
||
// Zuerst: ohne lesbaren Master-Key und Deployment-Key kann der Worker nichts tun, und das
|
||
// soll beim Start auffallen, nicht beim ersten Vorgang.
|
||
await masterKeyPruefen(await deploymentKeySicherstellen());
|
||
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);
|
||
}
|
||
}
|
||
},
|
||
);
|
||
|
||
// Nicht abgeholte Private Keys (#014) verfallen nach zwei Minuten. Ein Takt im Worker statt
|
||
// eines pg-boss-Zeitplans: dessen Raster ist eine Minute, und ein Fehler hier soll den
|
||
// Worker nicht anhalten. Zusaetzlich raeumt jede neue Abholung vorher auf, und das Web
|
||
// gibt eine abgelaufene Abholung ohnehin nicht mehr heraus.
|
||
const aufraeumen = () =>
|
||
abholungenAufraeumen().catch((fehler: unknown) =>
|
||
console.error(
|
||
"Aufräumen der Abholungen fehlgeschlagen:",
|
||
fehler instanceof Error ? fehler.message : fehler,
|
||
),
|
||
);
|
||
await aufraeumen();
|
||
const takt = setInterval(() => void aufraeumen(), 30_000);
|
||
|
||
// Periodischer Abgleich (#021). Der Zeitplan liegt in pg-boss und damit in der Datenbank:
|
||
// laufen spaeter mehrere Worker, feuert er trotzdem nur einmal.
|
||
const zeitplan = process.env["SSHKV_SYNC_CRON"]?.trim() || "30 2 * * *";
|
||
await boss.createQueue(ZEITPLAN_ABGLEICH);
|
||
await boss.schedule(ZEITPLAN_ABGLEICH, zeitplan, null, { tz: "Europe/Berlin" });
|
||
await boss.work(ZEITPLAN_ABGLEICH, async () => {
|
||
const anzahl = await alleHostsAbgleichen(null);
|
||
console.log(`Zeitplan: ${anzahl} Abgleich(e) eingereiht.`);
|
||
});
|
||
|
||
// Benachrichtigungen (#028): taeglich die Zusammenfassung, alle fuenf Minuten geaenderte
|
||
// Host-Keys (dringend) und faellige Wiederholungen. Fehler nur ins Log – das
|
||
// Versandprotokoll haelt fest, was nachzuholen ist.
|
||
const meldung = process.env["SSHKV_MELDUNG_CRON"]?.trim() || "0 7 * * *";
|
||
const melden = (lauf: () => Promise<{ gesendet: number; fehlgeschlagen: number }>) => async () => {
|
||
const bilanz = await lauf().catch((fehler: unknown) => {
|
||
console.error("Benachrichtigung:", fehler instanceof Error ? fehler.message : fehler);
|
||
return null;
|
||
});
|
||
if (bilanz && bilanz.gesendet + bilanz.fehlgeschlagen > 0)
|
||
console.log(`Benachrichtigung: ${bilanz.gesendet} gesendet, ${bilanz.fehlgeschlagen} fehlgeschlagen.`);
|
||
};
|
||
await boss.createQueue(ZEITPLAN_MELDUNG);
|
||
await boss.schedule(ZEITPLAN_MELDUNG, meldung, null, { tz: "Europe/Berlin" });
|
||
await boss.work(
|
||
ZEITPLAN_MELDUNG,
|
||
melden(() => zusammenfassungVersenden()),
|
||
);
|
||
await boss.createQueue(ZEITPLAN_MELDUNG_TAKT);
|
||
await boss.schedule(ZEITPLAN_MELDUNG_TAKT, "*/5 * * * *", null, { tz: "Europe/Berlin" });
|
||
await boss.work(
|
||
ZEITPLAN_MELDUNG_TAKT,
|
||
melden(() => taktVersenden()),
|
||
);
|
||
|
||
console.log(`sshkv-worker ${VERSION} bereit (Abgleich: ${zeitplan}, Meldungen: ${meldung}).`);
|
||
|
||
const beenden = async (signal: string) => {
|
||
console.log(`${signal} empfangen, beende laufende Vorgänge …`);
|
||
clearInterval(takt);
|
||
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);
|
||
});
|