Laufzettel/src/server/lauf-ereignisse.ts

import { Client } from "pg";
import { KANAL } from "@/lib/sse";

/**
 * Aenderungen an einem Lauf, wie die Datenbank sie meldet (#033).
 *
 * Die Trigger aus `20261005090000_lauf_ereignisse` schicken
 * `NOTIFY lauf_geaendert, '<laufId>'`. Hier haengt **eine** Verbindung je
 * Prozess mit `LISTEN` daran und verteilt an die offenen Stroeme.
 *
 * **Eine Verbindung, nicht eine je Strom.** `LISTEN` belegt eine Verbindung
 * dauerhaft; eine je Lasche waere bei vierzig Laschen vierzig belegte
 * Datenbankverbindungen, und PostgreSQL erlaubt voreingestellt hundert
 * insgesamt. Die Nutzlast sagt, welcher Lauf gemeint ist — das Verteilen
 * passiert hier im Prozess.
 *
 * **Nicht aus dem Prisma-Pool.** Ein Pool gibt Verbindungen zurueck und
 * verteilt sie neu; ein `LISTEN` darauf gilt nach der Rueckgabe fuer eine
 * fremde Anfrage oder fuer niemanden. Deshalb ein eigener `pg.Client`.
 */

type Rueckruf = () => void;

type Register = {
  readonly abonnenten: Map<string, Set<Rueckruf>>;
  verbindung: Client | undefined;
  aufbau: Promise<void> | undefined;
  geschlossen: boolean;
};

/*
 * An `globalThis`, aus demselben Grund wie der Prisma-Client in `src/lib/db.ts`:
 * Hot Reload laedt das Modul neu und baute sonst bei jedem Speichern eine
 * weitere LISTEN-Verbindung auf, ohne die alte zu schliessen.
 */
const global = globalThis as typeof globalThis & { laufzettelEreignisse?: Register };

function register(): Register {
  global.laufzettelEreignisse ??= {
    abonnenten: new Map(),
    verbindung: undefined,
    aufbau: undefined,
    geschlossen: false,
  };
  return global.laufzettelEreignisse;
}

/** Alle Abonnenten eines Laufs wecken. */
function wecken(laufId: string): void {
  for (const rueckruf of register().abonnenten.get(laufId) ?? []) {
    try {
      rueckruf();
    } catch (fehler) {
      // Ein kaputter Abonnent darf die anderen nicht mitnehmen.
      console.error("Abonnent eines Lauf-Ereignisses hat geworfen:", fehler);
    }
  }
}

/** Alle Abonnenten wecken, egal welcher Lauf. Nach einem Verbindungsabbruch. */
function alleWecken(): void {
  for (const laufId of [...register().abonnenten.keys()]) wecken(laufId);
}

async function verbindungAufbauen(): Promise<void> {
  const eigen = register();
  const url = process.env["DATABASE_URL"];
  if (!url) throw new Error("DATABASE_URL ist nicht gesetzt. Siehe .env.example.");

  const client = new Client({ connectionString: url });

  /*
   * **Erst der Fehlerbehandler, dann verbinden.** Ein `error` auf einem Client
   * ohne Behandler beendet in Node den ganzen Prozess — und eine Datenbank, die
   * neu startet, ist der Normalfall, nicht die Ausnahme.
   */
  client.on("error", (fehler) => {
    console.error("LISTEN-Verbindung verloren:", fehler.message);
    eigen.verbindung = undefined;
    void wiederverbinden();
  });
  client.on("end", () => {
    if (eigen.verbindung === client) {
      eigen.verbindung = undefined;
      void wiederverbinden();
    }
  });
  client.on("notification", (nachricht) => {
    if (nachricht.channel === KANAL && nachricht.payload) wecken(nachricht.payload);
  });

  await client.connect();
  await client.query(`LISTEN ${KANAL}`);
  eigen.verbindung = client;

  /*
   * **Nach einem Aufbau alle wecken.** Zwischen Abbruch und Wiederverbindung
   * sind Meldungen verloren gegangen — NOTIFY hat keinen Zwischenspeicher. Wer
   * jetzt nicht nachfragt, sieht den Haken des Kollegen nie. Lieber ein Abruf
   * zu viel als ein stiller Stand, der falsch ist.
   */
  alleWecken();
}

/**
 * Nach einem Abbruch neu verbinden, mit wachsendem Abstand.
 *
 * Ohne Abstand liefe bei einer laenger weg gebliebenen Datenbank eine
 * Endlosschleife, die nichts tut als Verbindungen zu versuchen.
 */
async function wiederverbinden(versuch = 1): Promise<void> {
  const eigen = register();
  if (eigen.geschlossen || eigen.abonnenten.size === 0) return;

  const wartezeit = Math.min(30_000, 500 * 2 ** (versuch - 1));
  await new Promise((fertig) => setTimeout(fertig, wartezeit));
  if (eigen.geschlossen || eigen.abonnenten.size === 0) return;

  try {
    await verbindungAufbauen();
  } catch {
    await wiederverbinden(versuch + 1);
  }
}

/** Sicherstellen, dass die eine Verbindung steht. Mehrfachaufrufe teilen sich den Aufbau. */
async function verbindungSicherstellen(): Promise<void> {
  const eigen = register();
  eigen.geschlossen = false;
  if (eigen.verbindung) return;
  // Laufender Aufbau: anhaengen statt einen zweiten starten.
  eigen.aufbau ??= verbindungAufbauen().finally(() => {
    eigen.aufbau = undefined;
  });
  await eigen.aufbau;
}

/**
 * Auf Aenderungen eines Laufs hoeren.
 *
 * Gibt die Funktion zurueck, mit der das Abonnement endet. Sie **muss** gerufen
 * werden, wenn der Strom schliesst — sonst waechst das Register mit jeder
 * geschlossenen Lasche weiter, und der Zaehler fuer die Obergrenze erzaehlt
 * Unsinn.
 */
export async function aufLaufHoeren(laufId: string, rueckruf: Rueckruf): Promise<() => void> {
  const eigen = register();
  const menge = eigen.abonnenten.get(laufId) ?? new Set<Rueckruf>();
  menge.add(rueckruf);
  eigen.abonnenten.set(laufId, menge);

  try {
    await verbindungSicherstellen();
  } catch (fehler) {
    // Aufraeumen, sonst zaehlt ein nie aufgebauter Strom weiter mit.
    menge.delete(rueckruf);
    if (menge.size === 0) eigen.abonnenten.delete(laufId);
    throw fehler;
  }

  let beendet = false;
  return () => {
    if (beendet) return;
    beendet = true;
    menge.delete(rueckruf);
    if (menge.size === 0) eigen.abonnenten.delete(laufId);
  };
}

/**
 * Zahl der offenen Stroeme in diesem Prozess.
 *
 * Grundlage der Obergrenze. Bewusst die Summe der Abonnenten und nicht die Zahl
 * der Laeufe: zwei Techniker an einem Lauf sind zwei Verbindungen.
 */
export function stroemeOffen(): number {
  let summe = 0;
  for (const menge of register().abonnenten.values()) summe += menge.size;
  return summe;
}

/**
 * Die Verbindung schliessen und alle Abonnements verwerfen.
 *
 * Fuer den geordneten Abbau und fuer Tests, die nicht auf einer offenen
 * Verbindung sitzen bleiben sollen.
 */
export async function ereignisseSchliessen(): Promise<void> {
  const eigen = register();
  eigen.geschlossen = true;
  eigen.abonnenten.clear();
  const client = eigen.verbindung;
  eigen.verbindung = undefined;
  if (client) await client.end().catch(() => undefined);
}