This commit is contained in:
Nick Szaunig committed
commit
59bb5a38a3
132 files changed
+19435
No files matched your search
@@ -0,0 +1,298 @@
|
||||
/**
|
||||
* Hintergrundaufgaben, ausgefuehrt vom Kubernetes-CronJob (scripts/worker.ts).
|
||||
*
|
||||
* Kein In-Process-Timer: Faelligkeiten stehen in der Datenbank, jeder Lauf
|
||||
* arbeitet den aktuellen Stand ab. Mehrere gleichzeitige Laeufe (z. B. ein
|
||||
* manueller Start waehrend des CronJobs) sind unschaedlich, weil jede Rate
|
||||
* vor dem Einzug atomar beansprucht wird und Stripe-Aufrufe
|
||||
* Idempotenzschluessel tragen.
|
||||
*/
|
||||
import { audit, WORKER } from "./audit";
|
||||
import type { Db } from "./db";
|
||||
import { flushEmails } from "./email";
|
||||
import { MAX_OFF_SESSION_ATTEMPTS, applyCheckoutExpired, applyFirstPaymentFailed, applyInstallmentFailed, applyInstallmentPaid, type EventContext } from "./payment-events";
|
||||
import type { PaymentGateway } from "./payment-gateway";
|
||||
import { purgeRateLimits } from "./rate-limit";
|
||||
import { settleCheckoutSession, type HandlerContext } from "./webhook";
|
||||
|
||||
/** Stripe-Idempotenzschluessel gelten 24 Stunden; ein haengender Einzug
|
||||
* wird nur innerhalb dieser Frist automatisch mit demselben Schluessel
|
||||
* wiederholt, danach entscheidet ein Mensch. */
|
||||
const SAFE_RETRY_WINDOW_MS = 20 * 60 * 60_000;
|
||||
const STUCK_AFTER_MS = 2 * 60 * 60_000;
|
||||
|
||||
function eventCtx(ctx: HandlerContext): EventContext {
|
||||
return { ...ctx, actor: WORKER };
|
||||
}
|
||||
|
||||
/**
|
||||
* Gibt abgelaufene Reservierungen frei. Vorher wird immer Stripe gefragt:
|
||||
* wurde die Session doch bezahlt (Webhook verloren), wird die Zahlung
|
||||
* verbucht statt der Platz freigegeben.
|
||||
*/
|
||||
export async function releaseExpiredReservations(db: Db, gateway: PaymentGateway, ctx: HandlerContext) {
|
||||
const expired = await db.seatReservation.findMany({
|
||||
where: { status: "ACTIVE", expiresAt: { lt: ctx.now } },
|
||||
include: { booking: { select: { id: true, livemode: true, stripeCheckoutSessionId: true } } },
|
||||
take: 100,
|
||||
});
|
||||
let released = 0;
|
||||
let settled = 0;
|
||||
for (const reservation of expired) {
|
||||
const booking = reservation.booking;
|
||||
const sameMode = booking.livemode === (gateway.mode === "live");
|
||||
if (booking.stripeCheckoutSessionId && sameMode) {
|
||||
let session = await gateway.retrieveCheckoutSession(booking.stripeCheckoutSessionId);
|
||||
if (session.status === "open") {
|
||||
await gateway.expireCheckoutSession(session.id);
|
||||
session = await gateway.retrieveCheckoutSession(session.id);
|
||||
}
|
||||
if (session.status === "complete") {
|
||||
await settleCheckoutSession(db, gateway, session, ctx, WORKER);
|
||||
settled++;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
const result = await db.$transaction((tx) =>
|
||||
applyCheckoutExpired(tx, booking.id, null, { ...eventCtx(ctx), reason: "reservation_expired" }),
|
||||
);
|
||||
if (result === "released") released++;
|
||||
}
|
||||
return { released, settled };
|
||||
}
|
||||
|
||||
/** Zieht faellige Folgeraten off-session ein. */
|
||||
export async function chargeDueInstallments(db: Db, gateway: PaymentGateway, ctx: HandlerContext) {
|
||||
const livemode = gateway.mode === "live";
|
||||
const due = await db.installment.findMany({
|
||||
where: {
|
||||
seq: { gt: 1 },
|
||||
booking: { status: "CONFIRMED", livemode },
|
||||
OR: [
|
||||
{ status: "PENDING", dueAt: { lte: ctx.now } },
|
||||
{ status: "FAILED", nextAttemptAt: { lte: ctx.now }, attemptCount: { lt: MAX_OFF_SESSION_ATTEMPTS } },
|
||||
],
|
||||
AND: [{ OR: [{ checkoutExpiresAt: null }, { checkoutExpiresAt: { lt: ctx.now } }] }],
|
||||
},
|
||||
orderBy: { dueAt: "asc" },
|
||||
take: 50,
|
||||
});
|
||||
|
||||
const summary = { charged: 0, failed: 0, requiresAction: 0, processing: 0, skipped: 0 };
|
||||
for (const installment of due) {
|
||||
// Atomar beanspruchen: nur ein Lauf darf diese Rate einziehen.
|
||||
const claimed = await db.installment.updateMany({
|
||||
where: {
|
||||
id: installment.id,
|
||||
status: installment.status,
|
||||
OR: [{ checkoutExpiresAt: null }, { checkoutExpiresAt: { lt: ctx.now } }],
|
||||
},
|
||||
data: { status: "PROCESSING", lastAttemptAt: ctx.now },
|
||||
});
|
||||
if (claimed.count === 0) {
|
||||
summary.skipped++;
|
||||
continue;
|
||||
}
|
||||
const booking = await db.booking.findUniqueOrThrow({ where: { id: installment.bookingId } });
|
||||
|
||||
// Schutz gegen jede Abbuchung ueber den vereinbarten Gesamtpreis hinaus,
|
||||
// insbesondere nach der letzten Rate.
|
||||
if (booking.paidCents + installment.amountCents > booking.totalCents) {
|
||||
await db.installment.update({ where: { id: installment.id }, data: { status: "CANCELED", failureCode: "plan_already_paid" } });
|
||||
await audit(db, WORKER, "installment.skipped_overpayment_guard", "Installment", installment.id, {
|
||||
paidCents: booking.paidCents,
|
||||
totalCents: booking.totalCents,
|
||||
});
|
||||
summary.skipped++;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!booking.stripeCustomerId || !booking.stripePaymentMethodId || booking.paymentMethodDetachedAt) {
|
||||
await db.$transaction((tx) =>
|
||||
applyInstallmentFailed(tx, installment.id, { paymentIntentId: null, failureCode: "no_saved_payment_method", requiresAction: true }, eventCtx(ctx)),
|
||||
);
|
||||
summary.requiresAction++;
|
||||
continue;
|
||||
}
|
||||
|
||||
let result;
|
||||
try {
|
||||
result = await gateway.chargeOffSession(
|
||||
{
|
||||
customerId: booking.stripeCustomerId,
|
||||
paymentMethodId: booking.stripePaymentMethodId,
|
||||
amountCents: installment.amountCents,
|
||||
description: `ARTISTA MODELCAMP 2027 · ${booking.reference} · Rate ${installment.seq}`,
|
||||
metadata: {
|
||||
kind: "installment_offsession",
|
||||
bookingId: booking.id,
|
||||
installmentId: installment.id,
|
||||
reference: booking.reference,
|
||||
},
|
||||
},
|
||||
// Gleicher Schluessel bei Wiederholung nach Netzwerkfehler: Stripe
|
||||
// liefert dann das erste Ergebnis statt ein zweites Mal abzubuchen.
|
||||
`offsession:${installment.id}:${installment.attemptCount}`,
|
||||
);
|
||||
} catch (error) {
|
||||
console.error(`[worker] Einzug Rate ${installment.id} unterbrochen: ${error instanceof Error ? error.message : "unbekannt"}`);
|
||||
await db.installment.updateMany({
|
||||
where: { id: installment.id, status: "PROCESSING" },
|
||||
data: { status: installment.status, nextAttemptAt: new Date(ctx.now.getTime() + 60 * 60_000) },
|
||||
});
|
||||
summary.failed++;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (result.outcome === "succeeded" && result.paymentIntentId) {
|
||||
await db.$transaction((tx) =>
|
||||
applyInstallmentPaid(
|
||||
tx,
|
||||
installment.id,
|
||||
{
|
||||
paymentIntentId: result.paymentIntentId!,
|
||||
chargeId: result.chargeId,
|
||||
paymentMethodId: booking.stripePaymentMethodId,
|
||||
customerId: booking.stripeCustomerId,
|
||||
amountCents: installment.amountCents,
|
||||
},
|
||||
eventCtx(ctx),
|
||||
),
|
||||
);
|
||||
summary.charged++;
|
||||
} else if (result.outcome === "processing") {
|
||||
await db.installment.update({ where: { id: installment.id }, data: { stripePaymentIntentId: result.paymentIntentId } });
|
||||
summary.processing++;
|
||||
} else {
|
||||
await db.$transaction((tx) =>
|
||||
applyInstallmentFailed(
|
||||
tx,
|
||||
installment.id,
|
||||
{
|
||||
paymentIntentId: result.paymentIntentId,
|
||||
failureCode: result.failureCode,
|
||||
requiresAction: result.outcome === "requires_action",
|
||||
},
|
||||
eventCtx(ctx),
|
||||
),
|
||||
);
|
||||
if (result.outcome === "requires_action") summary.requiresAction++;
|
||||
else summary.failed++;
|
||||
}
|
||||
}
|
||||
return summary;
|
||||
}
|
||||
|
||||
/** Faengt haengende Einzuege und verlorene Webhooks auf. */
|
||||
export async function reconcileStuck(db: Db, gateway: PaymentGateway, ctx: HandlerContext) {
|
||||
const stuck = await db.installment.findMany({
|
||||
where: { status: "PROCESSING", lastAttemptAt: { lt: new Date(ctx.now.getTime() - STUCK_AFTER_MS) }, seq: { gt: 1 } },
|
||||
include: { booking: { select: { livemode: true } } },
|
||||
take: 50,
|
||||
});
|
||||
let resolved = 0;
|
||||
for (const installment of stuck) {
|
||||
if (installment.booking.livemode !== (gateway.mode === "live")) continue;
|
||||
if (installment.stripePaymentIntentId) {
|
||||
const pi = await gateway.retrievePaymentIntent(installment.stripePaymentIntentId);
|
||||
if (pi.status === "succeeded") {
|
||||
await db.$transaction((tx) =>
|
||||
applyInstallmentPaid(
|
||||
tx,
|
||||
installment.id,
|
||||
{ paymentIntentId: pi.id, chargeId: pi.chargeId, paymentMethodId: pi.paymentMethodId, customerId: null, amountCents: pi.amountReceived },
|
||||
eventCtx(ctx),
|
||||
),
|
||||
);
|
||||
resolved++;
|
||||
} else if (pi.status === "requires_payment_method" || pi.status === "canceled" || pi.status === "requires_action") {
|
||||
await db.$transaction((tx) =>
|
||||
applyInstallmentFailed(
|
||||
tx,
|
||||
installment.id,
|
||||
{ paymentIntentId: pi.id, failureCode: `stuck_${pi.status}`, requiresAction: pi.status === "requires_action" },
|
||||
eventCtx(ctx),
|
||||
),
|
||||
);
|
||||
resolved++;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// Kein Ergebnis gespeichert (z. B. Absturz waehrend des Aufrufs).
|
||||
if (installment.lastAttemptAt && ctx.now.getTime() - installment.lastAttemptAt.getTime() < SAFE_RETRY_WINDOW_MS) {
|
||||
// attemptCount bleibt gleich -> naechster Lauf nutzt denselben
|
||||
// Idempotenzschluessel und kann nicht doppelt abbuchen.
|
||||
await db.installment.updateMany({
|
||||
where: { id: installment.id, status: "PROCESSING" },
|
||||
data: { status: "FAILED", nextAttemptAt: ctx.now, failureCode: "worker_interrupted" },
|
||||
});
|
||||
} else {
|
||||
await db.installment.updateMany({
|
||||
where: { id: installment.id, status: "PROCESSING" },
|
||||
data: { status: "REQUIRES_ACTION", failureCode: "manual_check_required" },
|
||||
});
|
||||
await audit(db, WORKER, "installment.manual_check_required", "Installment", installment.id);
|
||||
}
|
||||
resolved++;
|
||||
}
|
||||
|
||||
// Verzoegerte erste Zahlungen (SEPA), deren Ergebnis-Webhook ausblieb.
|
||||
const processing = await db.booking.findMany({
|
||||
where: {
|
||||
status: "PENDING",
|
||||
paymentStatus: "PROCESSING",
|
||||
livemode: gateway.mode === "live",
|
||||
updatedAt: { lt: new Date(ctx.now.getTime() - 24 * 60 * 60_000) },
|
||||
stripeCheckoutSessionId: { not: null },
|
||||
},
|
||||
take: 50,
|
||||
});
|
||||
for (const booking of processing) {
|
||||
const session = await gateway.retrieveCheckoutSession(booking.stripeCheckoutSessionId!);
|
||||
if (session.paymentStatus === "paid") {
|
||||
await settleCheckoutSession(db, gateway, session, ctx, WORKER);
|
||||
resolved++;
|
||||
} else if (session.paymentIntentId) {
|
||||
const pi = await gateway.retrievePaymentIntent(session.paymentIntentId);
|
||||
if (pi.status === "requires_payment_method" || pi.status === "canceled") {
|
||||
await db.$transaction((tx) => applyFirstPaymentFailed(tx, booking.id, eventCtx(ctx)));
|
||||
resolved++;
|
||||
}
|
||||
}
|
||||
}
|
||||
return { resolved };
|
||||
}
|
||||
|
||||
/**
|
||||
* Trennt die gespeicherte Zahlungsmethode, sobald keine Rate mehr offen ist
|
||||
* (vollstaendig bezahlt oder storniert). Danach ist technisch keine weitere
|
||||
* Abbuchung mehr moeglich.
|
||||
*/
|
||||
export async function detachFinishedPaymentMethods(db: Db, gateway: PaymentGateway, bookingIds?: string[]) {
|
||||
const candidates = await db.booking.findMany({
|
||||
where: {
|
||||
stripePaymentMethodId: { not: null },
|
||||
paymentMethodDetachedAt: null,
|
||||
livemode: gateway.mode === "live",
|
||||
...(bookingIds ? { id: { in: bookingIds } } : {}),
|
||||
installments: { none: { status: { in: ["PENDING", "PROCESSING", "FAILED", "REQUIRES_ACTION"] } } },
|
||||
},
|
||||
take: 50,
|
||||
});
|
||||
for (const booking of candidates) {
|
||||
await gateway.detachPaymentMethod(booking.stripePaymentMethodId!);
|
||||
await db.booking.update({ where: { id: booking.id }, data: { paymentMethodDetachedAt: new Date() } });
|
||||
await audit(db, WORKER, "payment_method.detached", "Booking", booking.id, { paymentMethodId: booking.stripePaymentMethodId });
|
||||
}
|
||||
return { detached: candidates.length };
|
||||
}
|
||||
|
||||
export async function runWorker(db: Db, gateway: PaymentGateway, ctx: HandlerContext) {
|
||||
const reservations = await releaseExpiredReservations(db, gateway, ctx);
|
||||
const reconcile = await reconcileStuck(db, gateway, ctx);
|
||||
const installments = await chargeDueInstallments(db, gateway, ctx);
|
||||
const detach = await detachFinishedPaymentMethods(db, gateway);
|
||||
const emails = await flushEmails(db, { limit: 200 });
|
||||
const purged = await purgeRateLimits(db);
|
||||
return { reservations, reconcile, installments, detach, emails, purgedRateLimits: purged };
|
||||
}
|
||||
Reference in new issue
Block a user