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"; /** Taegliche Zusammenfassung und Fuenf-Minuten-Takt der Benachrichtigungen (#028). */ export const ZEITPLAN_MELDUNG = "sshkv-meldung-zeitplan"; export const ZEITPLAN_MELDUNG_TAKT = "sshkv-meldung-takt"; /** 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 | 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 { const zwischenspeicher = globalThis as typeof globalThis & { sshkvBoss?: Promise }; 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 { const boss = await warteschlangeHolen(); return boss.send(WARTESCHLANGE, job); }