Ereignisse erst nach dem Commit ausliefern

Benachrichtigungs- und Mailversand lief im Handler noch in der offenen Transaktion und
zählte gegen deren Zeitlimit; unter Volllast brachen dadurch wechselnde Tests ab
(test-einsatz-field 31 s, test-planung-recommend). withDeferredEvents puffert Ereignisse
innerhalb von inTransaction und stellt sie nach dem Commit zu, bei Rollback gar nicht —
analog zu withDeferredAudit und passend zum dokumentierten Vertrag in events.ts.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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");
+});

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<T>(ctx: ServiceCtx, fn: () => Promise<T>): Promise<T> {
+  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<void> {
+  const buffer = deferredEvents.getStore();
+  if (buffer) {
+    buffer.items.push({ ctx, event });
+    return;
+  }
+  await deliver(ctx, event);
+}
+
+async function deliver(ctx: ServiceCtx, event: DomainEvent): Promise<void> {
   try {
     const { handleEvent } = await import("@/server/services/notifications/handle-event");
     await handleEvent(ctx, event);

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<T>(ctx: ServiceCtx, fn: (ctx: ServiceCtx) => Promise<T>): Promise<T> {
-  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 {