Terminplaner: Mail-Warteschlange mit Stundenbudget
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
ClaudeandClaude Sonnet 5.5 committed
1 parent
70b904260a
commit
af4029af81
2 files changed
+913
No files matched your search
@@ -0,0 +1,327 @@
|
||||
import { config } from '../../config.js'
|
||||
import { Query, createDocument, deleteDocument, getDocument, listAllDocuments, updateDocument } from '../appwriteAdmin.js'
|
||||
import { withLocks } from '../keyedLock.js'
|
||||
import { sendGuestMail } from '../mailer.js'
|
||||
import { renderMail } from '../mailTemplates.js'
|
||||
import { pruefeNow, pruefeText } from './util.js'
|
||||
|
||||
/**
|
||||
* Mail-Warteschlange des Terminplaners (termineOutbox; Spezifikation 2026-10-02-terminplaner-design.md, Abschnitt 6).
|
||||
* Alle Termine-Mails laufen hier durch: erst speichern (einreihen), dann senden (sendeEintrag, der einzige Sendepfad).
|
||||
*
|
||||
* Sendepfad (sendeEintrag): Pause und Stundenbudget pruefen, Sperre `termine-mail:<id>` (je Eintrag, parallele Laeufe
|
||||
* senden nie doppelt), Eintrag neu lesen, senden, Ergebnis speichern. Fehlerklassen:
|
||||
* - TypeError (ungueltige Adresse, fehlender Betreff ...) => sofort `fehler`, kein Versuch am Server
|
||||
* - EAUTH / SMTP 535 (Anmeldung) => 15 Minuten Pause fuer ALLE Eintraege; der Eintrag bleibt
|
||||
* `wartet`, auch `versuche` bleibt (kein Fehlversuch der Mail)
|
||||
* - sonstige 5xx => sofort `fehler`
|
||||
* - 4xx, Verbindungsfehler, nicht konfiguriertes SMTP => erneut nach 1, 5 und 30 Minuten, nach dem vierten
|
||||
* Fehlschlag `fehler`
|
||||
* Nach `gesendet` werden an, html, text und ics geleert (personenbezogene Daten).
|
||||
*
|
||||
* `naechsterVersuch` ist die Faelligkeit des Eintrags: bei `wartet` der naechste Sendeversuch, bei `gesendet` der
|
||||
* Loeschtermin (Versand + 7 Tage), bei `fehler` ebenfalls (Fehlschlag + 48 Stunden). So findet raeumeAuf alles ueber
|
||||
* den vorhandenen Index idx_status_next (status, naechsterVersuch); ein eigenes Fehlerdatum gibt es im Schema nicht.
|
||||
*
|
||||
* Stundenbudget (config.termine.mailsProStunde) und Anmeldepause sind Zustand dieses Prozesses: gleitendes Fenster der
|
||||
* Sendezeitpunkte bzw. ein Zeitpunkt. Ein Platz wird vor dem Senden belegt (gleichzeitige Versuche koennen das Budget
|
||||
* nicht ueberziehen) und bei einem Fehlschlag wieder freigegeben. `now` ist immer ein Parameter (feste Uhr in Tests);
|
||||
* ohne Angabe gilt bei einreihen und sendeEintrag die Wanduhr.
|
||||
*/
|
||||
|
||||
const col = () => config.collections.termineOutbox
|
||||
|
||||
const MIN_MS = 60 * 1000
|
||||
const STUNDE_MS = 60 * MIN_MS
|
||||
const TAG_MS = 24 * STUNDE_MS
|
||||
/** Wartezeit nach dem 1., 2. und 3. Fehlschlag; der vierte ist endgueltig. */
|
||||
const WIEDERHOLUNG_MS = [1 * MIN_MS, 5 * MIN_MS, 30 * MIN_MS]
|
||||
const ANMELDE_PAUSE_MS = 15 * MIN_MS
|
||||
const GESENDET_BEHALTEN_MS = 7 * TAG_MS
|
||||
const FEHLER_BEHALTEN_MS = 48 * STUNDE_MS
|
||||
const ICS_METHODEN = ['REQUEST', 'CANCEL', 'PUBLISH']
|
||||
/** Buchungs-IDs je Abfrage in zustellFehler (haelt die Adresse kurz). */
|
||||
const ID_BLOCK = 50
|
||||
|
||||
const iso = (wertMs) => new Date(wertMs).toISOString()
|
||||
|
||||
// Zustand des Prozesses
|
||||
let pauseBisMs = 0
|
||||
/** Zeitpunkte (ms) der Sendeversuche im letzten Stundenfenster (belegte Plaetze). */
|
||||
let sendeZeiten = []
|
||||
/** Laufende sendeEintrag-Aufrufe (fuer warteAufVersand). */
|
||||
const laufend = new Set()
|
||||
/** Eintraege, die schon rausgingen, deren Ergebnis aber nicht gespeichert werden konnte: id => Versandzeitpunkt (ms). */
|
||||
const unverbucht = new Map()
|
||||
|
||||
/** Anzahl freier Plaetze im Stundenbudget zum Zeitpunkt nowMs; raeumt abgelaufene Plaetze aus dem Fenster. */
|
||||
function freieBudgetPlaetze(nowMs) {
|
||||
const grenze = nowMs - STUNDE_MS
|
||||
sendeZeiten = sendeZeiten.filter((zeit) => zeit > grenze)
|
||||
return Math.max(0, config.termine.mailsProStunde - sendeZeiten.length)
|
||||
}
|
||||
|
||||
/** Pause oder ausgeschoepftes Budget (nur lesen). */
|
||||
const gebremst = (nowMs) => pauseBisMs > nowMs || freieBudgetPlaetze(nowMs) === 0
|
||||
|
||||
/** Belegt einen Platz im Budget (synchron, ohne Zwischenschritt); false bei Pause oder leerem Budget. */
|
||||
function belegePlatz(nowMs) {
|
||||
if (gebremst(nowMs)) return false
|
||||
sendeZeiten.push(nowMs)
|
||||
return true
|
||||
}
|
||||
|
||||
function gibPlatzFrei(nowMs) {
|
||||
const stelle = sendeZeiten.indexOf(nowMs)
|
||||
if (stelle >= 0) sendeZeiten.splice(stelle, 1)
|
||||
}
|
||||
|
||||
/** 'dauerhaft' | 'anmeldung' | 'vorlaeufig' (siehe Kopfkommentar). */
|
||||
function fehlerArt(err) {
|
||||
if (err instanceof TypeError) return 'dauerhaft'
|
||||
const antwort = Number(err?.responseCode)
|
||||
if (err?.code === 'EAUTH' || antwort === 535) return 'anmeldung'
|
||||
if (antwort >= 500 && antwort <= 599) return 'dauerhaft'
|
||||
return 'vorlaeufig'
|
||||
}
|
||||
|
||||
/** Kurzer Fehlertext fuers Feld `fehler` (hoechstens 300 Zeichen). */
|
||||
const fehlerText = (err) =>
|
||||
[err?.code, err?.responseCode, err?.message]
|
||||
.filter(Boolean)
|
||||
.join(' ')
|
||||
.replace(/\s+/g, ' ')
|
||||
.slice(0, 300)
|
||||
|
||||
/**
|
||||
* Mail einreihen: rendert die Vorlage sofort, speichert den Eintrag als `wartet` und stoesst den ersten Sendeversuch
|
||||
* an, ohne auf ihn zu warten (Fehler dabei stehen im Log, der Eintrag bleibt fuer verarbeite). Wirft bei ungueltigen
|
||||
* Eingaben (TypeError) oder wenn das Speichern scheitert; dann ist nichts gespeichert und nichts gesendet.
|
||||
* `ics` und `icsMethod` gehoeren zusammen: die METHOD-Zeile der Kalenderdatei muss icsMethod entsprechen.
|
||||
* @param {{ featureId: string, bookingId: string, art: string, an: string, fromName: string, replyTo?: string|null,
|
||||
* template: string, values?: object, blocks?: object, ics?: string|null, icsMethod?: 'REQUEST'|'CANCEL'|'PUBLISH' }} daten
|
||||
* @param {Date} [now]
|
||||
* @returns {Promise<object>} das gespeicherte Dokument (Status wartet)
|
||||
*/
|
||||
export async function einreihen(
|
||||
{ featureId, bookingId, art, an, fromName, replyTo = null, template, values = {}, blocks = {}, ics = null, icsMethod = 'REQUEST' },
|
||||
now = new Date(),
|
||||
) {
|
||||
for (const [wert, name] of [[featureId, 'featureId'], [bookingId, 'bookingId'], [art, 'art'], [an, 'an'], [fromName, 'fromName'], [template, 'template']]) {
|
||||
pruefeText(wert, name)
|
||||
}
|
||||
const nowMs = pruefeNow(now, 'einreihen')
|
||||
if (ics) {
|
||||
if (!ICS_METHODEN.includes(icsMethod)) throw new TypeError(`icsMethod muss ${ICS_METHODEN.join(', ')} sein`)
|
||||
if (!new RegExp(`^METHOD:${icsMethod}\\r?$`, 'm').test(ics)) throw new TypeError(`ics: METHOD passt nicht zu icsMethod ${icsMethod}`)
|
||||
}
|
||||
|
||||
const mail = renderMail(template, values, blocks)
|
||||
for (const problem of mail.problems) console.warn(`[termine-outbox] Vorlage ${template}: ${problem}`)
|
||||
|
||||
const doc = await createDocument(col(), {
|
||||
featureId,
|
||||
bookingId,
|
||||
art,
|
||||
an,
|
||||
fromName,
|
||||
replyTo: replyTo || '',
|
||||
subject: mail.subject,
|
||||
html: mail.html,
|
||||
text: mail.text,
|
||||
ics: ics || '',
|
||||
icsMethod: ics ? icsMethod : '',
|
||||
status: 'wartet',
|
||||
versuche: 0,
|
||||
naechsterVersuch: iso(nowMs),
|
||||
fehler: '',
|
||||
createdAt: iso(nowMs),
|
||||
})
|
||||
sendeEintrag(doc.$id, now).catch((err) => console.error(`[termine-outbox] Eintrag ${doc.$id}: ${err.message}`))
|
||||
return doc
|
||||
}
|
||||
|
||||
/**
|
||||
* Einziger Sendepfad. Macht hoechstens einen Sendeversuch fuer den Eintrag, wenn er `wartet`, faellig ist, keine Pause
|
||||
* laeuft und das Stundenbudget reicht.
|
||||
* @param {string} id
|
||||
* @param {Date} [now]
|
||||
* @returns {Promise<'gesendet'|'wartet'|'fehler'|'uebersprungen'>} gesendet = dieser Aufruf hat die Mail gesendet;
|
||||
* wartet = Eintrag bleibt in der Warteschlange (Pause, Budget, noch nicht faellig, vorlaeufiger Fehler); fehler = dieser
|
||||
* Aufruf hat ihn endgueltig scheitern lassen; uebersprungen = Eintrag fehlt oder ist schon gesendet/fehlgeschlagen
|
||||
*/
|
||||
export function sendeEintrag(id, now = new Date()) {
|
||||
const lauf = sende(id, now).finally(() => laufend.delete(lauf))
|
||||
laufend.add(lauf)
|
||||
return lauf
|
||||
}
|
||||
|
||||
async function sende(id, now) {
|
||||
const nowMs = pruefeNow(now, 'sendeEintrag')
|
||||
pruefeText(id, 'id')
|
||||
if (gebremst(nowMs)) return 'wartet'
|
||||
|
||||
return withLocks([`termine-mail:${id}`], async () => {
|
||||
let doc
|
||||
try {
|
||||
doc = await getDocument(col(), id)
|
||||
} catch (err) {
|
||||
if (err.status === 404) return 'uebersprungen'
|
||||
throw err
|
||||
}
|
||||
if (doc.status !== 'wartet') {
|
||||
unverbucht.delete(id)
|
||||
return 'uebersprungen'
|
||||
}
|
||||
if (unverbucht.has(id)) {
|
||||
await speichereGesendet(doc, unverbucht.get(id))
|
||||
return 'gesendet'
|
||||
}
|
||||
if (new Date(doc.naechsterVersuch).getTime() > nowMs) return 'wartet'
|
||||
// Der Platz wird ohne Zwischenschritt vor dem Senden belegt: gleichzeitige Versuche koennen das Budget nicht ueberziehen
|
||||
if (!belegePlatz(nowMs)) return 'wartet'
|
||||
|
||||
try {
|
||||
await sendGuestMail({
|
||||
to: doc.an,
|
||||
fromName: doc.fromName,
|
||||
replyTo: doc.replyTo || null,
|
||||
subject: doc.subject,
|
||||
html: doc.html,
|
||||
text: doc.text,
|
||||
ics: doc.ics || null,
|
||||
icsMethod: doc.icsMethod || 'REQUEST',
|
||||
})
|
||||
} catch (err) {
|
||||
gibPlatzFrei(nowMs)
|
||||
return speichereFehlschlag(doc, err, nowMs)
|
||||
}
|
||||
await speichereGesendet(doc, nowMs)
|
||||
return 'gesendet'
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Nach dem Versand: Status, Zeiten setzen und personenbezogene Felder leeren. Scheitert das Speichern (auch beim
|
||||
* zweiten Versuch), merkt sich der Prozess den Versand, damit der Eintrag nicht noch einmal gesendet wird.
|
||||
*/
|
||||
async function speichereGesendet(doc, gesendetMs) {
|
||||
const felder = {
|
||||
status: 'gesendet',
|
||||
versuche: (Number(doc.versuche) || 0) + 1,
|
||||
gesendetAt: iso(gesendetMs),
|
||||
naechsterVersuch: iso(gesendetMs + GESENDET_BEHALTEN_MS),
|
||||
fehler: '',
|
||||
an: '',
|
||||
html: '',
|
||||
text: '',
|
||||
ics: '',
|
||||
}
|
||||
let letzterFehler
|
||||
for (let versuch = 0; versuch < 2; versuch++) {
|
||||
try {
|
||||
await updateDocument(col(), doc.$id, felder)
|
||||
unverbucht.delete(doc.$id)
|
||||
return
|
||||
} catch (err) {
|
||||
letzterFehler = err
|
||||
}
|
||||
}
|
||||
unverbucht.set(doc.$id, gesendetMs)
|
||||
console.error(`[termine-outbox] Eintrag ${doc.$id} gesendet, Speichern gescheitert: ${letzterFehler.message}`)
|
||||
}
|
||||
|
||||
async function speichereFehlschlag(doc, err, nowMs) {
|
||||
const art = fehlerArt(err)
|
||||
if (art === 'anmeldung') {
|
||||
pauseBisMs = Math.max(pauseBisMs, nowMs + ANMELDE_PAUSE_MS)
|
||||
console.error(`[termine-outbox] SMTP-Anmeldung fehlgeschlagen (${err.code || err.responseCode}), Warteschlange pausiert bis ${iso(pauseBisMs)}`)
|
||||
return 'wartet'
|
||||
}
|
||||
const versuche = (Number(doc.versuche) || 0) + 1
|
||||
const wartezeit = WIEDERHOLUNG_MS[versuche - 1]
|
||||
if (art === 'vorlaeufig' && wartezeit !== undefined) {
|
||||
await updateDocument(col(), doc.$id, { versuche, naechsterVersuch: iso(nowMs + wartezeit), fehler: fehlerText(err) })
|
||||
return 'wartet'
|
||||
}
|
||||
await updateDocument(col(), doc.$id, { status: 'fehler', versuche, naechsterVersuch: iso(nowMs + FEHLER_BEHALTEN_MS), fehler: fehlerText(err) })
|
||||
console.error(`[termine-outbox] Eintrag ${doc.$id} endgueltig gescheitert nach ${versuche} Versuch(en): ${err.code || ''} ${err.responseCode || ''}`.trim())
|
||||
return 'fehler'
|
||||
}
|
||||
|
||||
/**
|
||||
* Minutenjob: sendet alle faelligen wartenden Eintraege, aelteste zuerst, bis Pause oder Stundenbudget bremsen.
|
||||
* Fehler bei einzelnen Eintraegen (z. B. Appwrite) stehen im Log und stoppen den Lauf nicht.
|
||||
* @param {Date} now
|
||||
* @returns {Promise<{ faellig: number, gesendet: number }>}
|
||||
*/
|
||||
export async function verarbeite(now) {
|
||||
const nowMs = pruefeNow(now, 'verarbeite')
|
||||
if (pauseBisMs > nowMs) return { faellig: 0, gesendet: 0 }
|
||||
const faellig = await listAllDocuments(col(), [Query.equal('status', 'wartet'), Query.lessThanEqual('naechsterVersuch', iso(nowMs))])
|
||||
faellig.sort((a, b) => new Date(a.createdAt) - new Date(b.createdAt) || (a.$id < b.$id ? -1 : 1))
|
||||
let gesendet = 0
|
||||
for (const eintrag of faellig) {
|
||||
if (gebremst(nowMs)) break
|
||||
try {
|
||||
if ((await sendeEintrag(eintrag.$id, now)) === 'gesendet') gesendet += 1
|
||||
} catch (err) {
|
||||
console.error(`[termine-outbox] Eintrag ${eintrag.$id}: ${err.message}`)
|
||||
}
|
||||
}
|
||||
return { faellig: faellig.length, gesendet }
|
||||
}
|
||||
|
||||
/** Nur fuer Tests: wartet, bis alle laufenden Sendeversuche (auch die aus einreihen) beendet sind. */
|
||||
export async function warteAufVersand() {
|
||||
while (laufend.size > 0) await Promise.allSettled([...laufend])
|
||||
}
|
||||
|
||||
/**
|
||||
* Buchungen (aus bookingIds) mit mindestens einer endgueltig gescheiterten Mail (Status fehler). Das Portal zeigt dort
|
||||
* „Bestätigung konnte nicht zugestellt werden“. Die Eintraege verschwinden mit raeumeAuf (48 Stunden nach dem Fehler).
|
||||
* @param {string[]} bookingIds
|
||||
* @returns {Promise<Set<string>>}
|
||||
*/
|
||||
export async function zustellFehler(bookingIds) {
|
||||
const ids = [...new Set((bookingIds ?? []).filter((id) => typeof id === 'string' && id !== ''))]
|
||||
const betroffen = new Set()
|
||||
for (let ab = 0; ab < ids.length; ab += ID_BLOCK) {
|
||||
const eintraege = await listAllDocuments(col(), [Query.equal('bookingId', ids.slice(ab, ab + ID_BLOCK)), Query.equal('status', 'fehler')])
|
||||
for (const eintrag of eintraege) betroffen.add(eintrag.bookingId)
|
||||
}
|
||||
return betroffen
|
||||
}
|
||||
|
||||
/**
|
||||
* Taeglicher Job: loescht gesendete Eintraege 7 Tage nach dem Versand und fehlgeschlagene 48 Stunden nach dem Fehler
|
||||
* (beides steht in naechsterVersuch). Wartende Eintraege bleiben.
|
||||
* @param {Date} now
|
||||
* @returns {Promise<number>} Anzahl geloeschter Eintraege
|
||||
*/
|
||||
export async function raeumeAuf(now) {
|
||||
const nowMs = pruefeNow(now, 'raeumeAuf')
|
||||
const alt = await listAllDocuments(col(), [Query.equal('status', ['gesendet', 'fehler']), Query.lessThanEqual('naechsterVersuch', iso(nowMs))])
|
||||
let geloescht = 0
|
||||
for (const eintrag of alt) {
|
||||
try {
|
||||
await deleteDocument(col(), eintrag.$id)
|
||||
geloescht += 1
|
||||
} catch (err) {
|
||||
if (err.status !== 404) throw err // 404: ein paralleler Lauf war schneller
|
||||
}
|
||||
}
|
||||
return geloescht
|
||||
}
|
||||
|
||||
/** Ende der letzten Anmeldepause (Date) oder null, wenn es noch keine gab. Eine Pause gilt, solange das Ende nach `now` liegt. */
|
||||
export function pausiertBis() {
|
||||
return pauseBisMs ? new Date(pauseBisMs) : null
|
||||
}
|
||||
|
||||
/** Nur fuer Tests: Pause, Budgetfenster und nicht gespeicherte Versände zuruecksetzen. */
|
||||
export function resetOutboxForTests() {
|
||||
pauseBisMs = 0
|
||||
sendeZeiten = []
|
||||
unverbucht.clear()
|
||||
}
|
||||
Reference in new issue
Block a user