FireAdministration/src/worker/index.ts

// 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 { atemschutzprotokollErzeugen } from "@/lib/atemschutz/protokoll-erzeugen";
import { erinnerungenSenden, mangelBenachrichtigung, terminePlanen } from "@/worker/wartung";
import {
  benachrichtigungenAufraeumen,
  morgenlauf,
  zustellen,
} from "@/worker/benachrichtigung";
import { diktatErkennen, veralteteSeitenEinbetten, wikiEinbetten } from "@/worker/ki";

/**
 * 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<AuftragsDaten>) {
  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<AuftragsDaten>) {
  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<AuftragsDaten>) {
  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<AuftragsDaten>) {
  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 atemschutzprotokollErzeugenAuftrag(auftrag: Job<AuftragsDaten>) {
  const { organisationId, ueberwachungId } = auftrag.data as AuftragsDaten & {
    ueberwachungId?: string;
  };

  if (!ueberwachungId) {
    throw new Error("Protokollauftrag ohne Angabe, wozu — ueberwachungId fehlt.");
  }

  const ergebnis = await atemschutzprotokollErzeugen(organisationId, ueberwachungId);

  if (!ergebnis) {
    protokoll(
      `Kein Atemschutzprotokoll für ${ueberwachungId} — unbekannt oder noch nicht beendet.`,
    );
    return { erzeugt: false };
  }

  protokoll(
    ergebnis.neu
      ? `Atemschutzprotokoll für ${ueberwachungId} erzeugt.`
      : `Atemschutzprotokoll für ${ueberwachungId} lag bereits vor.`,
  );

  return { erzeugt: ergebnis.neu, dateiId: ergebnis.dateiId };
}

/**
 * Der Zustelllauf.
 *
 * Läuft jede Minute und ist deshalb bewusst wortkarg: er meldet sich nur,
 * wenn er etwas getan hat. Ein Worker, der 1440-mal am Tag „nichts zu tun"
 * ins Protokoll schreibt, macht das Protokoll unlesbar.
 */
async function benachrichtigungZustellen(auftrag: Job<AuftragsDaten>) {
  const ergebnis = await zustellen(auftrag.data.organisationId);
  if (ergebnis.gesendet + ergebnis.aufgegeben > 0) {
    protokoll(
      `Benachrichtigungen: ${ergebnis.gesendet} zugestellt, ${ergebnis.aufgegeben} aufgegeben`,
    );
  }
  return ergebnis;
}

/** Der Morgenlauf: Terminerinnerungen, ablaufende Qualifikationen, Zustellung. */
async function benachrichtigungErinnern(auftrag: Job<AuftragsDaten>) {
  const ergebnis = await morgenlauf(auftrag.data.organisationId);
  protokoll(
    `Morgenlauf: ${ergebnis.termine.termine} Termine, ` +
      `${ergebnis.qualifikationen.qualifikationen} Qualifikationen, ` +
      `${ergebnis.zugestellt.gesendet} Nachrichten zugestellt`,
  );
  return ergebnis;
}

/**
 * Eine Wiki-Seite durchsuchbar machen.
 *
 * Im Hintergrund, weil eine lange Seite zwanzig Modellaufrufe bedeutet. Bis
 * es durch ist, bleibt der alte Suchstand nutzbar — und die Seitenansicht
 * sagt, dass er veraltet ist.
 */
async function wikiEinbettenAuftrag(auftrag: Job<AuftragsDaten>) {
  const { organisationId, seiteId } = auftrag.data as AuftragsDaten & { seiteId?: string };

  if (!seiteId) {
    throw new Error("Einbettungsauftrag ohne Angabe, wofür — seiteId fehlt.");
  }

  return wikiEinbetten(organisationId, seiteId);
}

/** Eine Diktataufnahme in Text verwandeln. */
async function diktatErkennenAuftrag(auftrag: Job<AuftragsDaten>) {
  const { organisationId, diktatId } = auftrag.data as AuftragsDaten & { diktatId?: string };

  if (!diktatId) {
    throw new Error("Diktatauftrag ohne Angabe, wofür — diktatId fehlt.");
  }

  return diktatErkennen(organisationId, diktatId);
}

async function aufbewahrungPruefen(auftrag: Job<AuftragsDaten>): Promise<{ entfernt: number }> {
  protokoll(`Aufbewahrungsfristen prüfen für Organisation ${auftrag.data.organisationId}`);

  // Seit Phase 7: zugestellte Benachrichtigungen nach ihrer Frist abräumen.
  // Fehlgeschlagene bleiben — sie zeigen einen kaputten Kanal an.
  const benachrichtigungen = await benachrichtigungenAufraeumen(auftrag.data.organisationId);

  // Seit Phase 8: Wiki-Seiten nachziehen, deren Suchstand veraltet ist. Fängt
  // den Fall ab, dass der Worker stand oder das Modell nicht erreichbar war,
  // als jemand gespeichert hat.
  await veralteteSeitenEinbetten(auftrag.data.organisationId).catch((fehler) => {
    console.error("[worker] Nachträgliches Einbetten fehlgeschlagen:", fehler);
  });

  // Ab Phase 8: Audit-Log exportieren und danach abräumen,
  // ausgetretene Mitglieder nach PERSON_ANONYMISIERUNG_TAGE anonymisieren.
  return { entfernt: benachrichtigungen.entfernt };
}

const bearbeiter: Record<string, (auftrag: Job<AuftragsDaten>) => Promise<unknown>> = {
  [AUFTRAG.FRISTEN_PRUEFEN]: fristenPruefen,
  [AUFTRAG.ERINNERUNG_SENDEN]: erinnerungSenden,
  [AUFTRAG.PDF_ERZEUGEN]: pdfErzeugen,
  [AUFTRAG.EINSATZBERICHT_ERZEUGEN]: einsatzberichtErzeugenAuftrag,
  [AUFTRAG.ATEMSCHUTZPROTOKOLL_ERZEUGEN]: atemschutzprotokollErzeugenAuftrag,
  [AUFTRAG.BENACHRICHTIGUNG_ZUSTELLEN]: benachrichtigungZustellen,
  [AUFTRAG.BENACHRICHTIGUNG_ERINNERN]: benachrichtigungErinnern,
  [AUFTRAG.WIKI_EINBETTEN]: wikiEinbettenAuftrag,
  [AUFTRAG.DIKTAT_ERKENNEN]: diktatErkennenAuftrag,
  [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<number> {
  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,
    );
    // Jede Minute. Eine Absage am Dienstagabend muss ankommen, solange sie
    // noch etwas nützt — nicht am nächsten Morgen.
    await beauftrageWiederkehrend(
      AUFTRAG.BENACHRICHTIGUNG_ZUSTELLEN,
      { organisationId: organisation.id },
      "* * * * *",
    );
    await beauftrageWiederkehrend(
      AUFTRAG.BENACHRICHTIGUNG_ERINNERN,
      { organisationId: organisation.id },
      env.BENACHRICHTIGUNG_LAUF,
    );
  }

  return organisationen.length;
}

const worker = new Worker<AuftragsDaten>(
  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<void> {
  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"));