make it work with mikro-orm .transactional usage? (vibe-kanban 3c966397)

which relies on async local storage

see https://mikro-orm.io/docs/transactions

and `create-tour-participant.ts` in `~/src/tournee/be`

app code would look like this

\\`\`\`

  @transactional()
  private async createParticipant(
    auth: AuthContext,
    division: TourDivisionEntity,
    input: CreateTourParticipantInput,
  ) {
    const nextNumber = await this.getNextParticipantNumber(division);
    const participant = await this.createSingleParticipant(
      auth,
      division,
      input,
      nextNumber,
    );

    await this.emitter.emitAsync(
      TourParticipantsCreated.name,
      new TourParticipantsCreated([participant]),
    );

    return participant;
  }
\\`\`\`

can this lib work with a code such as the one above ? (p.s. how event object, name nad payload is created, that's a detail - dont focus on this now)

Author: jtomaszewski

packages/core/src/emitter/transactional-event-emitter.ts

@@ -78,7 +78,7 @@ export class TransactionalEventEmitter {
     entities: {
       operation: TransactionalEventEmitterOperations;
       entity: object;
-    }[],
+    }[] = [],
     customDatabaseDriverPersister?: DatabaseDriverPersister,
   ): Promise<void> {
     return this.emitInternal(event, entities, customDatabaseDriverPersister, false);
@@ -89,7 +89,7 @@ export class TransactionalEventEmitter {
     entities: {
       operation: TransactionalEventEmitterOperations;
       entity: object;
-    }[],
+    }[] = [],
     customDatabaseDriverPersister?: DatabaseDriverPersister,
   ): Promise<void> {
     return this.emitInternal(event, entities, customDatabaseDriverPersister, true);

packages/mikroorm-driver/README.md

@@ -109,6 +109,11 @@ new MikroORMDatabaseDriverFactory(orm, {
 new MikroORMDatabaseDriverFactory(orm, {
   listenNotify: { enabled: false },
 })
+
+// Enable AsyncLocalStorage context (for @Transactional() decorator)
+new MikroORMDatabaseDriverFactory(orm, {
+  useContext: true,
+})
 ```
 
 ### How It Works
@@ -119,6 +124,55 @@ The `PostgreSQLEventListener`:
 - Works alongside polling as a fallback mechanism
 - Requires the LISTEN/NOTIFY migration from `InboxOutboxMigrations`
 
+## @Transactional() Decorator Support
+
+The driver supports MikroORM's `@Transactional()` decorator via AsyncLocalStorage context propagation. Enable `useContext: true` to participate in the same transaction:
+
+```typescript
+// Module setup
+InboxOutboxModule.registerAsync({
+  imports: [MikroOrmModule],
+  useFactory: (orm: MikroORM) => ({
+    driverFactory: new MikroORMDatabaseDriverFactory(orm, {
+      useContext: true, // Enable context propagation
+    }),
+    events: [...],
+    retryEveryMilliseconds: 30_000,
+    maxInboxOutboxTransportEventPerRetry: 10,
+  }),
+  inject: [MikroORM],
+}),
+```
+
+```typescript
+// Service using @Transactional()
+import { Transactional } from '@mikro-orm/core';
+
+@Injectable()
+export class OrderService {
+  constructor(
+    private em: EntityManager,
+    private emitter: TransactionalEventEmitter,
+  ) {}
+
+  @Transactional()
+  async createOrder(data: CreateOrderInput) {
+    const order = this.em.create(Order, data);
+    this.em.persist(order);
+
+    // Event is persisted in the same transaction
+    await this.emitter.emitAsync(new OrderCreatedEvent(order.id));
+
+    return order;
+  }
+}
+```
+
+With `useContext: true`:
+- Event persistence participates in the `@Transactional()` transaction
+- Rollbacks affect both business data and events atomically
+- No `em.clear()` after flush (preserves shared identity map)
+
 ## Supported Databases
 
 | Database   | Real-time Support |

packages/mikroorm-driver/src/driver/mikroorm-database-driver.factory.ts

@@ -5,10 +5,18 @@ import { MikroORMDatabaseDriver } from './mikroorm.database-driver';
 
 export interface MikroORMDatabaseDriverFactoryOptions {
   listenNotify?: PostgreSQLEventListenerOptions & { enabled?: boolean };
+  /**
+   * When true, uses the current transactional context (via AsyncLocalStorage)
+   * instead of forking a new EntityManager. This enables compatibility with
+   * MikroORM's @Transactional() decorator.
+   * @default false
+   */
+  useContext?: boolean;
 }
 
 export class MikroORMDatabaseDriverFactory {
   private eventListener: PostgreSQLEventListener | null = null;
+  private useContext: boolean;
 
   constructor(
     private readonly orm: MikroORM,
@@ -17,11 +25,14 @@ export class MikroORMDatabaseDriverFactory {
     if (options?.listenNotify?.enabled !== false) {
       this.eventListener = new PostgreSQLEventListener(orm, options?.listenNotify);
     }
+    this.useContext = options?.useContext ?? false;
   }
 
   create(eventConfigurationResolver: EventConfigurationResolverContract): DatabaseDriver {
-    const forkedEm = this.orm.em.fork();
-    return new MikroORMDatabaseDriver(forkedEm, eventConfigurationResolver);
+    const em = this.useContext ? this.orm.em.getContext() : this.orm.em.fork();
+    return new MikroORMDatabaseDriver(em, eventConfigurationResolver, {
+      clearAfterFlush: !this.useContext,
+    });
   }
 
   getEventListener(): EventListener | null {

packages/mikroorm-driver/src/driver/mikroorm.database-driver.ts

@@ -1,10 +1,20 @@
 import { EntityManager, LockMode } from '@mikro-orm/core';
 import { DatabaseDriver, EventConfigurationResolverContract, InboxOutboxTransportEvent } from '@nestixis/nestjs-inbox-outbox';
 import { MikroOrmInboxOutboxTransportEvent } from '../model/mikroorm-inbox-outbox-transport-event.model';
-  
+
+export interface MikroORMDatabaseDriverOptions {
+  clearAfterFlush?: boolean;
+}
+
 export class MikroORMDatabaseDriver implements DatabaseDriver {
+  private readonly clearAfterFlush: boolean;
+
   constructor(
-    private readonly em: EntityManager, private readonly eventConfigurationResolver: EventConfigurationResolverContract) {
+    private readonly em: EntityManager,
+    private readonly eventConfigurationResolver: EventConfigurationResolverContract,
+    options?: MikroORMDatabaseDriverOptions,
+  ) {
+    this.clearAfterFlush = options?.clearAfterFlush ?? true;
   }
 
   async findAndExtendReadyToRetryEvents(limit: number): Promise<InboxOutboxTransportEvent[]> {
@@ -40,7 +50,9 @@ export class MikroORMDatabaseDriver implements DatabaseDriver {
 
   async flush(): Promise<void> {
     await this.em.flush();
-    this.em.clear();
+    if (this.clearAfterFlush) {
+      this.em.clear();
+    }
   }
 
   createInboxOutboxTransportEvent(eventName: string, eventPayload: any, expireAt: number, readyToRetryAfter: number | null): InboxOutboxTransportEvent {

packages/mikroorm-driver/src/test/mikroorm-database-driver-factory.spec.ts

@@ -118,6 +118,40 @@ describe('MikroORMDatabaseDriverFactory', () => {
     });
   });
 
+  describe('useContext option', () => {
+    it('should use forked EntityManager when useContext is false (default)', () => {
+      const factory = new MikroORMDatabaseDriverFactory(orm);
+      const driver = factory.create(createEventConfigResolver());
+
+      expect(driver).toBeInstanceOf(MikroORMDatabaseDriver);
+    });
+
+    it('should use context EntityManager when useContext is true', () => {
+      const factory = new MikroORMDatabaseDriverFactory(orm, {
+        useContext: true,
+      });
+      const driver = factory.create(createEventConfigResolver());
+
+      expect(driver).toBeInstanceOf(MikroORMDatabaseDriver);
+    });
+
+    it('should support combining useContext with listenNotify options', () => {
+      const factory = new MikroORMDatabaseDriverFactory(orm, {
+        useContext: true,
+        listenNotify: {
+          channelName: 'custom_channel',
+        },
+      });
+
+      const driver = factory.create(createEventConfigResolver());
+      const listener = factory.getEventListener() as PostgreSQLEventListener;
+
+      expect(driver).toBeInstanceOf(MikroORMDatabaseDriver);
+      expect(listener).toBeInstanceOf(PostgreSQLEventListener);
+      expect(listener.channelName).toBe('custom_channel');
+    });
+  });
+
   describe('getEventListener', () => {
     it('should return PostgreSQLEventListener by default', () => {
       const factory = new MikroORMDatabaseDriverFactory(orm);

packages/mikroorm-driver/src/test/test-utils.ts

@@ -20,6 +20,7 @@ export interface TestAppConfig {
   retryEveryMilliseconds?: number;
   maxInboxOutboxTransportEventPerRetry?: number;
   databaseType?: DatabaseType;
+  useContext?: boolean;
 }
 
 export interface TestContext {
@@ -112,7 +113,9 @@ export async function createTestApp(config: TestAppConfig): Promise<TestContext>
   const inboxOutboxModule = InboxOutboxModule.registerAsync({
     imports: [MikroOrmModule],
     useFactory: (orm: MikroORM) => {
-      const driverFactory = new MikroORMDatabaseDriverFactory(orm);
+      const driverFactory = new MikroORMDatabaseDriverFactory(orm, {
+        useContext: config.useContext,
+      });
       return {
         driverFactory,
         events: config.events,