// Im Container liefert Compose die Umgebung, lokal die .env. dotenv // überschreibt nichts, was schon gesetzt ist — im Betrieb ändert die Zeile // also nichts, macht aber "npm run worker" auf dem Entwicklungsrechner möglich. import "dotenv/config"; import { Worker, type Job } from "bullmq"; import { env } from "@/lib/env"; import { AUFTRAG, WARTESCHLANGE, beauftrageWiederkehrend, redis, type AuftragsDaten, } from "@/lib/warteschlange"; import { ohneMandantenfilter, verbindungenTrennen } from "@/lib/db"; import { protokollErzeugen } from "@/lib/wartung/protokoll-erzeugen"; import { einsatzberichtErzeugen } from "@/lib/einsatz/bericht-erzeugen"; import { erinnerungenSenden, mangelBenachrichtigung, terminePlanen } from "@/worker/wartung"; /** * Hintergrundprozess. * * Läuft als eigener Container neben der Anwendung und arbeitet die * Warteschlange ab. Seit Phase 2 steckt die eigentliche Arbeit drin: * Terminplanung, Erinnerungen und die Erzeugung der Prüfprotokolle, ab Phase 3 * dazu die Einsatzberichte. Die Aufbewahrung folgt in Phase 8. * * Beim Start trägt sich der Prozess die wiederkehrenden Läufe je Wehr ein. * Das ist bewusst hier und nicht in der Anwendung: der Worker ist der Einzige, * der sie auch ausführt, und ein Zeitplan ohne Ausführenden ist nutzlos. */ function protokoll(nachricht: string, ...weiteres: unknown[]): void { console.info(`[worker] ${nachricht}`, ...weiteres); } async function fristenPruefen(auftrag: Job) { const { organisationId } = auftrag.data; const ergebnis = await terminePlanen(organisationId); protokoll( `Fristen geprüft für ${organisationId}: ${ergebnis.angelegt} Termine geplant, ` + `${ergebnis.statusGeaendert} Geräte im Status geändert`, ); return ergebnis; } async function erinnerungSenden(auftrag: Job) { const { organisationId, anlass, mangelId } = auftrag.data as AuftragsDaten & { anlass?: string; mangelId?: string; }; // Mängel melden sich sofort, Termine im nächtlichen Sammellauf. if (anlass?.startsWith("mangel") && mangelId) { return mangelBenachrichtigung(organisationId, mangelId, anlass); } return erinnerungenSenden(organisationId); } async function pdfErzeugen(auftrag: Job) { const { organisationId, pruefungId } = auftrag.data as AuftragsDaten & { pruefungId?: string }; if (!pruefungId) { throw new Error("PDF-Auftrag ohne Angabe, wozu — pruefungId fehlt."); } const ergebnis = await protokollErzeugen(organisationId, pruefungId); if (!ergebnis) { protokoll(`Kein Protokoll für Prüfung ${pruefungId} — unbekannt oder noch Entwurf.`); return { erzeugt: false }; } protokoll( ergebnis.neu ? `Protokoll für Prüfung ${pruefungId} erzeugt.` : `Protokoll für Prüfung ${pruefungId} lag bereits vor.`, ); return { erzeugt: ergebnis.neu, dateiId: ergebnis.dateiId }; } async function einsatzberichtErzeugenAuftrag(auftrag: Job) { const { organisationId, einsatzId } = auftrag.data as AuftragsDaten & { einsatzId?: string }; if (!einsatzId) { throw new Error("Berichtsauftrag ohne Angabe, wozu — einsatzId fehlt."); } const ergebnis = await einsatzberichtErzeugen(organisationId, einsatzId); if (!ergebnis) { protokoll(`Kein Bericht für Einsatz ${einsatzId} — unbekannt oder nicht freigegeben.`); return { erzeugt: false }; } protokoll( ergebnis.neu ? `Einsatzbericht für ${einsatzId} erzeugt.` : `Einsatzbericht für ${einsatzId} lag bereits vor.`, ); return { erzeugt: ergebnis.neu, dateiId: ergebnis.dateiId }; } async function aufbewahrungPruefen(auftrag: Job): Promise<{ entfernt: number }> { protokoll(`Aufbewahrungsfristen prüfen für Organisation ${auftrag.data.organisationId}`); // Ab Phase 8: Audit-Log exportieren und danach abräumen, // ausgetretene Mitglieder nach PERSON_ANONYMISIERUNG_TAGE anonymisieren. return { entfernt: 0 }; } const bearbeiter: Record) => Promise> = { [AUFTRAG.FRISTEN_PRUEFEN]: fristenPruefen, [AUFTRAG.ERINNERUNG_SENDEN]: erinnerungSenden, [AUFTRAG.PDF_ERZEUGEN]: pdfErzeugen, [AUFTRAG.EINSATZBERICHT_ERZEUGEN]: einsatzberichtErzeugenAuftrag, [AUFTRAG.AUFBEWAHRUNG_PRUEFEN]: aufbewahrungPruefen, }; /** * Trägt die nächtlichen Läufe für jede aktive Wehr ein. * * BullMQ hält wiederkehrende Aufträge in Redis; ein zweiter Eintrag mit * derselben Auftragskennung ersetzt den ersten. Der Aufruf ist deshalb bei * jedem Start wiederholbar, und eine neu angelegte Wehr kommt beim nächsten * Neustart des Workers dazu. */ async function zeitplaeneEinrichten(): Promise { const organisationen = await ohneMandantenfilter( "Wartungsjobs für alle Wehren einplanen", (db) => db.organisation.findMany({ where: { aktiv: true, geloeschtAm: null }, select: { id: true, kurzname: true } }), ); for (const organisation of organisationen) { await beauftrageWiederkehrend( AUFTRAG.FRISTEN_PRUEFEN, { organisationId: organisation.id }, env.WARTUNG_NACHTLAUF, ); await beauftrageWiederkehrend( AUFTRAG.ERINNERUNG_SENDEN, { organisationId: organisation.id }, env.WARTUNG_ERINNERUNG_LAUF, ); } return organisationen.length; } const worker = new Worker( WARTESCHLANGE, async (auftrag) => { const bearbeiten = bearbeiter[auftrag.name]; if (!bearbeiten) { throw new Error(`Unbekannte Auftragsart: ${auftrag.name}`); } const beginn = Date.now(); const ergebnis = await bearbeiten(auftrag); protokoll(`${auftrag.name} erledigt in ${Date.now() - beginn} ms`); return ergebnis; }, { connection: redis(), concurrency: env.WORKER_CONCURRENCY, }, ); worker.on("failed", (auftrag, fehler) => { console.error( `[worker] Auftrag ${auftrag?.name ?? "?"} (${auftrag?.id ?? "?"}) fehlgeschlagen:`, fehler.message, ); }); worker.on("ready", () => { protokoll( `bereit — Warteschlange "${WARTESCHLANGE}", ${env.WORKER_CONCURRENCY} parallele Aufträge`, ); void zeitplaeneEinrichten() .then((anzahl) => protokoll( `Zeitpläne eingerichtet für ${anzahl} Wehren ` + `(Planung ${env.WARTUNG_NACHTLAUF}, Erinnerungen ${env.WARTUNG_ERINNERUNG_LAUF})`, ), ) .catch((fehler) => { // Ein fehlgeschlagener Zeitplan darf den Worker nicht aufhalten: die // Aufträge aus der Anwendung sollen weiter abgearbeitet werden. console.error("[worker] Zeitpläne ließen sich nicht einrichten:", fehler); }); }); /** Laufende Aufträge zu Ende bringen, bevor der Prozess endet. */ async function herunterfahren(signal: string): Promise { protokoll(`${signal} empfangen — fahre herunter ...`); await worker.close(); await verbindungenTrennen(); redis().disconnect(); protokoll("beendet."); process.exit(0); } process.on("SIGTERM", () => void herunterfahren("SIGTERM")); process.on("SIGINT", () => void herunterfahren("SIGINT"));