Transactionele events na database-commit

Stel: een medewerker wordt aangemeld voor een shift. Dat triggert een transactie: de toewijzing wordt opgeslagen, de beschikbaarheid bijgewerkt, en de medewerker ontvangt een notificatie. Die notificatie wordt gegenereerd door een event-handler die op ShiftAssigned reageert.

Als de event-handler draait voordat de transactie gecommit is, kan de database nog terugdraaien. De medewerker ontvangt dan een notificatie voor een toewijzing die nooit heeft plaatsgevonden.

Dit is het fundamentele probleem met event-buses en database-transacties: ze zijn onafhankelijke systemen die je expliciet moet synchroniseren.

De EnhancedEventBus

De oplossing is een event-bus die events buffert zolang een transactie open is:

@Injectable()
export class EnhancedEventBusService {
  private pendingTransactionEvents = new Map<string, PendingEvent[]>()

  async emit<T>(event: BaseEvent<T>, options?: EmitOptions): Promise<void> {
    if (options?.transactionId) {
      // Transactie open: buffer het event
      const pending = this.pendingTransactionEvents.get(options.transactionId) ?? []
      pending.push({ event, options })
      this.pendingTransactionEvents.set(options.transactionId, pending)
    } else {
      // Geen transactie: direct emiten
      await this.orchestrator.dispatch(event, options)
    }
  }

  async commitTransaction(transactionId: string): Promise<void> {
    const pending = this.pendingTransactionEvents.get(transactionId) ?? []
    this.pendingTransactionEvents.delete(transactionId)

    // Emit alle gebufferde events ná de DB-commit
    for (const { event, options } of pending) {
      await this.orchestrator.dispatch(event, { ...options, transactionId: undefined })
    }
  }

  rollbackTransaction(transactionId: string): void {
    // Gooi alle gebufferde events weg — de DB-wijzigingen bestaan niet
    this.pendingTransactionEvents.delete(transactionId)
  }
}

De Map<string, PendingEvent[]> indexed op transactie-ID. Meerdere transacties kunnen tegelijkertijd lopen — elke transactie heeft zijn eigen buffer.

commitTransaction emiteert de events ná het verwijderen uit de map. Dat is bewust: als het emiten mislukt, raakt de map niet corrupt.

Prisma middleware die de transactie-ID injecteert

De event-bus heeft een transactionId nodig. Maar in Prisma weet applicatiecode niet standaard of een operatie onderdeel is van een $transaction(). Dat lossen we op met een middleware die de transactie-scope detecteert:

export function createTransactionAwareMiddleware(
  eventBus: EnhancedEventBusService,
): Prisma.Middleware {
  return async (params, next) => {
    // params.runInTransaction == true EN depth 1 = begin van een nieuwe transactie
    const isTransactionRoot =
      params.runInTransaction && params.__internalParams?.txDepth === 1;

    if (!isTransactionRoot) {
      return next(params);
    }

    const transactionId = randomUUID();

    // Injecteer transactie-ID in alle sub-operaties via AsyncLocalStorage
    try {
      const result = await next(params);
      await eventBus.commitTransaction(transactionId);
      return result;
    } catch (err) {
      eventBus.rollbackTransaction(transactionId);
      throw err;
    }
  };
}

txDepth === 1 detecteert de buitenste transactie. Geneste writes in dezelfde transactie hebben een hogere depth en worden overgeslagen door de middleware — die registreren events via de AsyncLocalStorage-context die de buitenste middleware heeft gezet.

Op succes: commitTransaction → events worden emitted. Op fout: rollbackTransaction → events worden weggegooid, de exception bubbles up.

AsyncLocalStorage voor transactie-context

Het probleem met middleware: de transactionId moet beschikbaar zijn voor alle service-aanroepen die events emiten binnen die transactie — maar die services hebben geen directe toegang tot de middleware-context.

AsyncLocalStorage lost dat op:

const transactionStore = new AsyncLocalStorage<{ transactionId: string }>();

// In de middleware: context opzetten
return transactionStore.run({ transactionId }, () => next(params));

// In services die events emiten:
export class ShiftService {
  async assignShift(shiftId: string, userId: string): Promise<void> {
    // ... database write via Prisma ...
    const ctx = transactionStore.getStore();
    await this.eventBus.emit(new ShiftAssignedEvent(shiftId, userId), {
      transactionId: ctx?.transactionId, // undefined buiten transactie → direct emit
    });
  }
}

Services hoeven niet te weten of ze in een transactie zitten. Ze lezen de context uit de store — als die er is, wordt het event gebufferd; zo niet, dan wordt het direct emitted.

Event-fases: sync en async

Niet alle event-handlers moeten gebufferd worden. We onderscheiden twee fases:

BEFORE_RESPONSE — synchrone handlers die de response beïnvloeden (validatie, aanvullende writes). Deze draaien altijd in dezelfde aanroepketen en zijn automatisch transactie-aware.

AFTER_RESPONSE — asynchrone handlers (notificaties sturen, externe systemen informeren). Deze worden via BullMQ ingepland na commit:

@EventHandler(ShiftAssignedEvent, { phase: EventPhase.AFTER_RESPONSE })
export class SendShiftNotificationHandler {
  async handle(event: ShiftAssignedEvent): Promise<void> {
    await this.notificationService.send(event.userId, {
      type: 'shift_assigned',
      shiftId: event.shiftId,
    })
  }
}

De orchestrator plaatst AFTER_RESPONSE-handlers in BullMQ met attempts: 3 en exponential backoff starting bij 2 seconden. Als de handler crasht, probeert BullMQ het automatisch opnieuw — zonder dat de gebruiker iets merkt.

BEFORE_RESPONSE-handlers die blocking: true hebben, laten errors propageren naar de caller. Die kunnen de transactie laten terugdraaien — wat precies de bedoeling is voor validatielogica.

Wat dit patroon oplost en wat niet

Dit patroon garandeert dat events pas emitted worden ná een succesvolle DB-commit. Het lost niet op:

• Handler-failures ná commit: als een AFTER_RESPONSE-handler definitief faalt (na alle retries), is de database al gewijzigd maar het event niet verwerkt. BullMQ slaat mislukte jobs op in een dead-letter queue voor handmatig herstel.
• Distributed transactions: als je twee databases hebt, helpt dit patroon alleen voor de Prisma-database. Voor cross-service transacties heb je saga-patterns of outbox-patronen nodig.

Voor enkelvoudige-database applicaties is dit patroon solide en eenvoudig te begrijpen: events volgen altijd de DB, nooit andersom.