feat(orm): add adapter event dispatcher and logger

Author: marcj

packages/framework/src/module.ts

@@ -292,8 +292,9 @@ export class FrameworkModule extends createModuleClass({
     protected setupDatabase() {
         for (const db of this.dbs) {
             this.configureProvider<DatabaseRegistry>(v => v.addDatabase(db.classType, {}, db.module));
-            db.module.configureProvider((db: Database, eventDispatcher: EventDispatcher, stopwatch: Stopwatch) => {
-                db.eventDispatcher = eventDispatcher;
+            db.module.configureProvider((db: Database, eventDispatcher: EventDispatcher, logger: Logger, stopwatch: Stopwatch) => {
+                db.setEventDispatcher(eventDispatcher);
+                db.setLogger(logger);
                 db.stopwatch = stopwatch;
             }, {}, db.classType);
         }

packages/mongo/src/adapter.ts

@@ -27,13 +27,15 @@ import { MongoPersistence } from './persistence.js';
 import { MongoClient } from './client/client.js';
 import { DeleteCommand } from './client/command/delete.js';
 import { MongoQueryResolver } from './query.resolver.js';
-import { MongoDatabaseTransaction } from './client/connection.js';
+import { MongoDatabaseTransaction, MongoDatabaseTransactionMonitor } from './client/connection.js';
 import { CreateIndex, CreateIndexesCommand } from './client/command/createIndexes.js';
 import { DropIndexesCommand } from './client/command/dropIndexes.js';
 import { CreateCollectionCommand } from './client/command/createCollection.js';
 import { entity, ReceiveType, ReflectionClass, resolveReceiveType } from '@deepkit/type';
 import { Command } from './client/command/command.js';
 import { AggregateCommand } from './client/command/aggregate.js';
+import { EventDispatcher } from '@deepkit/event';
+import { Logger } from '@deepkit/logger';
 
 export class MongoDatabaseQueryFactory extends DatabaseAdapterQueryFactory {
     constructor(
@@ -114,6 +116,7 @@ export class MongoDatabaseAdapter extends DatabaseAdapter {
     public readonly client: MongoClient;
 
     protected ormSequences: ReflectionClass<any>;
+    protected transactionMonitor: MongoDatabaseTransactionMonitor = new MongoDatabaseTransactionMonitor(this.logger);
 
     constructor(
         connection: string | MongoClient,
@@ -130,6 +133,17 @@ export class MongoDatabaseAdapter extends DatabaseAdapter {
         this.ormSequences = ReflectionClass.from(OrmSequence);
     }
 
+    setEventDispatcher(eventDispatcher: EventDispatcher) {
+        super.setEventDispatcher(eventDispatcher);
+        this.client.setEventDispatcher(eventDispatcher);
+    }
+
+    setLogger(logger: Logger) {
+        super.setLogger(logger);
+        this.client.setLogger(logger);
+        this.transactionMonitor.logger = logger;
+    }
+
     rawFactory(session: DatabaseSession<this>): MongoRawFactory {
         return new MongoRawFactory(session, this.client);
     }
@@ -147,7 +161,7 @@ export class MongoDatabaseAdapter extends DatabaseAdapter {
     }
 
     createTransaction(session: DatabaseSession<this>): MongoDatabaseTransaction {
-        return new MongoDatabaseTransaction;
+        return new MongoDatabaseTransaction(this.transactionMonitor);
     }
 
     isNativeForeignKeyConstraintSupported() {

packages/mongo/src/client/connection.ts

@@ -24,7 +24,7 @@ import { AbortTransactionCommand } from './command/abortTransaction.js';
 import { DataEvent, EventDispatcher, EventTokenSync } from '@deepkit/event';
 import { IsMasterCommand } from './command/ismaster.js';
 import { Logger } from '@deepkit/logger';
-import { ConnectionOptions } from './options.js';
+import { CommandOptions, ConnectionOptions } from './options.js';
 
 export enum MongoConnectionStatus {
     pending = 'pending',
@@ -585,22 +585,56 @@ export function readUint32LE(buffer: Uint8Array | ArrayBuffer, offset: number =
     return buffer[offset] + (buffer[offset + 1] * 2 ** 8) + (buffer[offset + 2] * 2 ** 16) + (buffer[offset + 3] * 2 ** 24);
 }
 
+/**
+ * This class is responsible to monitor leaking transactions and automatically
+ * rollback them if the DatabaseSession was garbage collected without commit or rollback.
+ */
+export class MongoDatabaseTransactionMonitor {
+    finalizer = new FinalizationRegistry<{
+        connection: MongoConnection;
+        lsid: { id: string };
+        txnNumber: bigint;
+    }>((value) => {
+        this.logger.warn(`Leaking transaction detected. Automatically rollback transaction ${value.txnNumber}`);
+        value.connection.execute(new AbortTransactionCommand())
+            .catch((error) => {
+                this.logger.error(`Could not rollback transaction ${value.txnNumber}: ${formatError(error)}`);
+            }).finally(() => {
+            value.connection.release();
+        });
+    });
+
+    constructor(public logger: Logger) {
+    }
+}
+
 export class MongoDatabaseTransaction extends DatabaseTransaction {
+    commandOptions: CommandOptions = {};
     static txnNumber: bigint = 0n;
 
     connection?: MongoConnection;
     lsid?: { id: string };
     txnNumber: bigint = 0n;
     started: boolean = false;
 
+    constructor(protected monitor: MongoDatabaseTransactionMonitor) {
+        super();
+    }
+
+    with(commandOptions: CommandOptions): this {
+        Object.assign(this.commandOptions, commandOptions);
+        return this;
+    }
+
     applyTransaction(cmd: TransactionalMessage) {
-        if (!this.lsid) return;
+        if (!this.lsid || !this.connection) return;
         cmd.lsid = this.lsid;
         cmd.txnNumber = this.txnNumber;
         cmd.autocommit = false;
         if (!this.started && !cmd.abortTransaction && !cmd.commitTransaction) {
             this.started = true;
             cmd.startTransaction = true;
+            this.monitor.finalizer.register(this, { connection: this.connection, lsid: this.lsid, txnNumber: this.txnNumber });
         }
     }
 

packages/mongo/tests/mongo.spec.ts

@@ -16,13 +16,24 @@ import {
     UUID,
     uuid,
 } from '@deepkit/type';
-import { getInstanceStateFromItem, UniqueConstraintFailure } from '@deepkit/orm';
+import { Database, getInstanceStateFromItem, UniqueConstraintFailure } from '@deepkit/orm';
 import { SimpleModel, SuperSimple } from './entities.js';
 import { createDatabase } from './utils.js';
 import { databaseFactory } from './factory.js';
+import { MongoDatabaseAdapter } from '../src/adapter.js';
+import { MemoryLogger } from '@deepkit/logger';
 
 Error.stackTraceLimit = 100;
 
+test('logger', async () => {
+    const database = new Database(new MongoDatabaseAdapter('mongodb://invalid-host'));
+    const logger = new MemoryLogger();
+    database.setLogger(logger);
+    await expect(database.adapter.client.connect()).rejects.toThrow('getaddrinfo ENOTFOUND invalid-host');
+    expect(logger.memory.messageStrings[0]).toContain('getaddrinfo ENOTFOUND invalid-host');
+    database.disconnect();
+});
+
 test('test save undefined values', async () => {
     const session = await createDatabase('test save undefined values');
 

packages/mongo/tests/replica.spec.ts

@@ -82,7 +82,7 @@ describe('replica set, primary secondary', () => {
         expect(client.config.hosts[1].stats.commandsExecuted).toBe(3);
     });
 
-    test('transaction', async () => {
+    test('transaction write', async () => {
         const client = createClient(`mongodb://primary`);
         const database = new Database(new MongoDatabaseAdapter(client), [User]);
 
@@ -113,6 +113,21 @@ describe('replica set, primary secondary', () => {
             expect(await database.query(User).filter({ username: 'user1 changed' }).has()).toBe(true);
         }
     });
+
+    test('transaction read', async () => {
+        const client = createClient(`mongodb://primary`);
+        const database = new Database(new MongoDatabaseAdapter(client), [User]);
+
+        {
+            await database.query(User).deleteMany();
+            await database.persist(new User('user1'));
+
+            const session = database.createSession();
+            session.useTransaction();
+            const user = await session.query(User).findOne();
+            expect(user.username).toBe('user1');
+        }
+    });
 });
 
 describe.skip('local replica', () => {

packages/mongo/tsconfig.json

@@ -14,7 +14,8 @@
     "module": "CommonJS",
     "lib": [
       "es2020",
-      "es2022.error"
+      "es2022.error",
+      "es2021.weakref"
     ],
     "esModuleInterop": true,
     "outDir": "./dist/cjs",

packages/orm/src/database-adapter.ts

@@ -18,19 +18,11 @@ import {
     isClass,
     stringifyValueWithType,
 } from '@deepkit/core';
-import {
-    is,
-    isSameType,
-    ItemChanges,
-    PrimaryKeyFields,
-    ReceiveType,
-    ReflectionClass,
-    ReflectionKind,
-    stringifyType,
-    Type,
-} from '@deepkit/type';
+import { is, isSameType, ItemChanges, PrimaryKeyFields, ReceiveType, ReflectionClass, ReflectionKind, stringifyType, Type } from '@deepkit/type';
 import { Query } from './query.js';
 import { DatabaseSession, DatabaseTransaction } from './database-session.js';
+import { EventDispatcher } from '@deepkit/event';
+import { Logger } from '@deepkit/logger';
 
 export abstract class DatabaseAdapterQueryFactory {
     abstract createQuery<T extends OrmEntity>(type?: ReceiveType<T> | ClassType<T> | AbstractClassType<T> | ReflectionClass<T>): Query<T>;
@@ -109,6 +101,17 @@ export class MigrateOptions {
  * You can specify a more specialized adapter like MysqlDatabaseAdapter/MongoDatabaseAdapter with special API for MySQL/Mongo.
  */
 export abstract class DatabaseAdapter {
+    eventDispatcher: EventDispatcher = new EventDispatcher();
+    logger: Logger = new Logger();
+
+    setEventDispatcher(eventDispatcher: EventDispatcher) {
+        this.eventDispatcher = eventDispatcher;
+    }
+
+    setLogger(logger: Logger) {
+        this.logger = logger;
+    }
+
     abstract queryFactory(session: DatabaseSession<this>): DatabaseAdapterQueryFactory;
 
     rawFactory(session: DatabaseSession<this>): RawFactory<any> {

packages/orm/src/database.ts

@@ -32,6 +32,7 @@ import { Stopwatch } from '@deepkit/stopwatch';
 import { getClassState, getInstanceState, getNormalizedPrimaryKey } from './identity-map.js';
 import { EventDispatcher, EventDispatcherUnsubscribe, EventListenerCallback, EventToken } from '@deepkit/event';
 import { DatabasePlugin, DatabasePluginRegistry } from './plugin/plugin.js';
+import { Logger } from '@deepkit/logger';
 
 /**
  * Hydrates not completely populated item and makes it completely accessible.
@@ -174,6 +175,16 @@ export class Database<ADAPTER extends DatabaseAdapter = DatabaseAdapter> {
         }
     }
 
+    setEventDispatcher(eventDispatcher: EventDispatcher) {
+        this.eventDispatcher = eventDispatcher;
+        this.adapter.setEventDispatcher(eventDispatcher);
+    }
+
+    setLogger(logger: Logger) {
+        this.logger.logger = logger;
+        this.adapter.setLogger(logger);
+    }
+
     registerPlugin(...plugins: DatabasePlugin[]): void {
         for (const plugin of plugins) {
             this.pluginRegistry.add(plugin);