diff --git a/scripts/test-events-deferred.ts b/scripts/test-events-deferred.ts new file mode 100644 index 0000000..1287ef6 --- /dev/null +++ b/scripts/test-events-deferred.ts @@ -0,0 +1,70 @@ +// Ereignisse werden erst NACH dem Commit ausgeliefert (events.ts#withDeferredEvents, von +// services/context.ts#inTransaction gesetzt). Vorher lief der Benachrichtigungs- und Mailversand +// noch in der offenen Transaktion und zählte gegen deren Zeitlimit — unter Last brach der +// Einsatzstart deshalb sporadisch ab. +// (1) innerhalb der Transaktion: noch keine Benachrichtigung +// (2) nach dem Commit: Benachrichtigung vorhanden +// (3) Rollback: Ereignis wird verworfen, keine Benachrichtigung +// (4) ohne Transaktion: unverändert sofortige Auslieferung +// +// Lauf: npx tsx scripts/test-events-deferred.ts + +import "dotenv/config"; +import { randomUUID } from "node:crypto"; +import type { DomainEvent } from "../src/lib/events"; +import { prisma } from "../src/server/db"; +import { emitEvent } from "../src/server/events"; +import { inTransaction } from "../src/server/services/context"; +import { createTenant, ok, runSuite, section, type TenantFixture } from "./lib/e2e-fixture"; + +const SLUG = "zz-events-deferred"; +const TYPE = "work_order.assigned"; + +async function order(A: TenantFixture) { + return prisma.workOrder.create({ + data: { + tenantId: A.tenantId, + number: `EV-${randomUUID().slice(0, 6)}`, + title: "Heizung warten", + customerId: A.customerId, + siteId: A.siteId, + assignedTeamId: A.teamId, + status: "assigned", + }, + }); +} + +const ev = (entityId: string): DomainEvent => ({ type: TYPE, entityType: "work_order", entityId }); +const notifications = (tenantId: string, entityId: string) => prisma.notification.count({ where: { tenantId, type: TYPE, entityId } }); + +void runSuite("Ereignisse nach dem Commit", [SLUG], async () => { + const A = await createTenant(SLUG); + + section("Transaktion mit Commit"); + const wo1 = await order(A); + const duringTransaction = await inTransaction(A.ctx.backoffice, async (tx) => { + await emitEvent(tx, ev(wo1.id)); + return notifications(A.tenantId, wo1.id); + }); + ok(duringTransaction === 0, "(1) während der Transaktion wird noch nichts zugestellt"); + ok((await notifications(A.tenantId, wo1.id)) > 0, "(2) nach dem Commit ist die Benachrichtigung da"); + + section("Rollback"); + const wo2 = await order(A); + let rolledBack = false; + try { + await inTransaction(A.ctx.backoffice, async (tx) => { + await emitEvent(tx, ev(wo2.id)); + throw new Error("rollback"); + }); + } catch { + rolledBack = true; + } + ok(rolledBack, "(3) Transaktion abgebrochen"); + ok((await notifications(A.tenantId, wo2.id)) === 0, "(3) verworfenes Ereignis löst keine Benachrichtigung aus"); + + section("Ohne Transaktion"); + const wo3 = await order(A); + await emitEvent(A.ctx.backoffice, ev(wo3.id)); + ok((await notifications(A.tenantId, wo3.id)) > 0, "(4) ohne Transaktion unverändert sofort zugestellt"); +}); diff --git a/src/server/events.ts b/src/server/events.ts index f94da33..6cf7594 100644 --- a/src/server/events.ts +++ b/src/server/events.ts @@ -1,12 +1,44 @@ +import { AsyncLocalStorage } from "node:async_hooks"; import type { DomainEvent } from "@/lib/events"; import type { ServiceCtx } from "@/server/services/context"; +type Buffered = { ctx: ServiceCtx; event: DomainEvent }; +const deferredEvents = new AsyncLocalStorage<{ items: Buffered[] }>(); + +/** + * Collect events raised inside a transaction and deliver them AFTER the commit + * (services/context.ts#inTransaction). Handlers write notifications and queue mails; inside the + * transaction that work counted against the transaction timeout and could exceed it under load. + * On rollback the buffered events are dropped — nothing happened, so nothing is announced. + * Nested calls join the outer buffer. + */ +export async function withDeferredEvents(ctx: ServiceCtx, fn: () => Promise): Promise { + if (deferredEvents.getStore()) return fn(); + const buffer: { items: Buffered[] } = { items: [] }; + const result = await deferredEvents.run(buffer, fn); + for (const { ctx: eventCtx, event } of buffer.items) { + // the transaction client is closed after the commit → deliver with the caller's client + await deliver({ ...eventCtx, db: ctx.db }, event); + } + return result; +} + /** * Emit a domain event AFTER the mutation succeeded. Never throws: a failing notification * must not roll back business data — failures are logged and surfaced via sync/notification * monitoring. Lanes call ONLY this function; the notifications lane owns the handler. + * Inside `inTransaction` the event is buffered and delivered after the commit. */ export async function emitEvent(ctx: ServiceCtx, event: DomainEvent): Promise { + const buffer = deferredEvents.getStore(); + if (buffer) { + buffer.items.push({ ctx, event }); + return; + } + await deliver(ctx, event); +} + +async function deliver(ctx: ServiceCtx, event: DomainEvent): Promise { try { const { handleEvent } = await import("@/server/services/notifications/handle-event"); await handleEvent(ctx, event); diff --git a/src/server/services/context.ts b/src/server/services/context.ts index 05c2d84..c9e6757 100644 --- a/src/server/services/context.ts +++ b/src/server/services/context.ts @@ -1,6 +1,7 @@ import type { Session } from "next-auth"; import { tenantTransaction, type TenantDb } from "@/server/db"; import { withDeferredAudit } from "@/server/audit"; +import { withDeferredEvents } from "@/server/events"; /** * Context passed to every domain service. Created by server actions (from moduleGuard) @@ -29,10 +30,12 @@ export function ctxFromGuard(g: { session: Session; db: TenantDb; permissions: R * is bound to the transaction; nested calls join the outer transaction. * Never use `ctx.db.$transaction(...)` directly — it is not atomic with RLS_ENFORCED=true. * Audit entries written inside `fn` are flushed after the commit and dropped on rollback - * (except `denied`), see audit.ts#withDeferredAudit. + * (except `denied`), see audit.ts#withDeferredAudit. Domain events raised inside `fn` are delivered + * after the commit as well (events.ts#withDeferredEvents): notification and mail work must not + * count against the transaction timeout. */ export function inTransaction(ctx: ServiceCtx, fn: (ctx: ServiceCtx) => Promise): Promise { - return withDeferredAudit(() => tenantTransaction(ctx.db, ctx.tenantId, (tx) => fn({ ...ctx, db: tx }))); + return withDeferredEvents(ctx, () => withDeferredAudit(() => tenantTransaction(ctx.db, ctx.tenantId, (tx) => fn({ ...ctx, db: tx })))); } export function can(ctx: ServiceCtx, permission: string): boolean {