feat: impl dal transaction

Author: killagu

core/dal-runtime/src/MySqlDataSource.ts

@@ -43,4 +43,8 @@ export class MysqlDataSource extends Base {
   async query<T = any>(sql: string): Promise<T> {
     return this.client.query(sql);
   }
+
+  async beginTransactionScope<T>(scope: () => Promise<T>): Promise<T> {
+    return await this.client.beginTransactionScope(scope);
+  }
 }

plugin/dal/app.ts

@@ -6,11 +6,13 @@ import { MysqlDataSourceManager } from './lib/MysqlDataSourceManager';
 import { SqlMapManager } from './lib/SqlMapManager';
 import { TableModelManager } from './lib/TableModelManager';
 import { DalModuleLoadUnitHook } from './lib/DalModuleLoadUnitHook';
+import { TransactionPrototypeHook } from './lib/TransactionPrototypeHook';
 
 export default class ControllerAppBootHook {
   private readonly app: Application;
   private dalTableEggPrototypeHook: DalTableEggPrototypeHook;
   private dalModuleLoadUnitHook: DalModuleLoadUnitHook;
+  private transactionPrototypeHook: TransactionPrototypeHook;
 
   constructor(app: Application) {
     this.app = app;
@@ -19,7 +21,9 @@ export default class ControllerAppBootHook {
   configWillLoad() {
     this.dalModuleLoadUnitHook = new DalModuleLoadUnitHook(this.app.config.env, this.app.moduleConfigs);
     this.dalTableEggPrototypeHook = new DalTableEggPrototypeHook(this.app.logger);
+    this.transactionPrototypeHook = new TransactionPrototypeHook(this.app.moduleConfigs, this.app.logger);
     this.app.eggPrototypeLifecycleUtil.registerLifecycle(this.dalTableEggPrototypeHook);
+    this.app.eggPrototypeLifecycleUtil.registerLifecycle(this.transactionPrototypeHook);
     this.app.loadUnitLifecycleUtil.registerLifecycle(this.dalModuleLoadUnitHook);
   }
 
@@ -30,6 +34,9 @@ export default class ControllerAppBootHook {
     if (this.dalModuleLoadUnitHook) {
       this.app.loadUnitLifecycleUtil.deleteLifecycle(this.dalModuleLoadUnitHook);
     }
+    if (this.transactionPrototypeHook) {
+      this.app.eggPrototypeLifecycleUtil.deleteLifecycle(this.transactionPrototypeHook);
+    }
     MysqlDataSourceManager.instance.clear();
     SqlMapManager.instance.clear();
     TableModelManager.instance.clear();

plugin/dal/lib/TransactionPrototypeHook.ts

@@ -0,0 +1,58 @@
+import assert from 'assert';
+import { LifecycleHook, ModuleConfigHolder, Logger } from '@eggjs/tegg';
+import { EggPrototype, EggPrototypeLifecycleContext } from '@eggjs/tegg-metadata';
+import { PropagationType, TransactionMetaBuilder } from '@eggjs/tegg/transaction';
+import { Pointcut } from '@eggjs/tegg/aop';
+import { TransactionalAOP, TransactionalParams } from './TransactionalAOP';
+import { MysqlDataSourceManager } from './MysqlDataSourceManager';
+
+export class TransactionPrototypeHook implements LifecycleHook<EggPrototypeLifecycleContext, EggPrototype> {
+  private readonly moduleConfigs: Record<string, ModuleConfigHolder>;
+  private readonly logger: Logger;
+
+  constructor(moduleConfigs: Record<string, ModuleConfigHolder>, logger: Logger) {
+    this.moduleConfigs = moduleConfigs;
+    this.logger = logger;
+  }
+
+  public async preCreate(ctx: EggPrototypeLifecycleContext): Promise<void> {
+    const builder = new TransactionMetaBuilder(ctx.clazz);
+    const transactionMetadataList = builder.build();
+    if (transactionMetadataList.length < 1) {
+      return;
+    }
+    const moduleName = ctx.loadUnit.name;
+    for (const transactionMetadata of transactionMetadataList) {
+      const clazzName = `${moduleName}.${ctx.clazz.name}.${String(transactionMetadata.method)}`;
+      const datasourceConfigs = (this.moduleConfigs[moduleName]?.config as any)?.dataSource || {};
+
+      let datasourceName: string;
+      if (transactionMetadata.datasourceName) {
+        assert(datasourceConfigs[transactionMetadata.datasourceName], `method ${clazzName} specified datasource ${transactionMetadata.datasourceName} not exists`);
+        datasourceName = transactionMetadata.datasourceName;
+        this.logger.info(`use datasource [${transactionMetadata.datasourceName}] for class ${clazzName}`);
+      } else {
+        const dataSources = Object.keys(datasourceConfigs);
+        if (dataSources.length === 1) {
+          datasourceName = dataSources[0];
+        } else {
+          throw new Error(`method ${clazzName} not specified datasource, module ${moduleName} has multi datasource, should specify datasource name`);
+        }
+        this.logger.info(`use default datasource ${dataSources[0]} for class ${clazzName}`);
+      }
+      const adviceParams: TransactionalParams = {
+        propagation: transactionMetadata.propagation,
+        dataSourceGetter: () => {
+          const mysqlDataSource = MysqlDataSourceManager.instance.get(moduleName, datasourceName);
+          if (!mysqlDataSource) {
+            throw new Error(`method ${clazzName} not found datasource ${datasourceName}`);
+          }
+          return mysqlDataSource;
+        },
+      };
+      assert(adviceParams.propagation === PropagationType.REQUIRED, 'Transactional propagation only support required for now');
+      Pointcut(TransactionalAOP, { adviceParams })((ctx.clazz as any).prototype, transactionMetadata.method);
+    }
+  }
+
+}

plugin/dal/lib/TransactionalAOP.ts

@@ -0,0 +1,23 @@
+import { Advice, AdviceContext, IAdvice } from '@eggjs/tegg/aop';
+import { AccessLevel, EggProtoImplClass, ObjectInitType } from '@eggjs/tegg';
+import { PropagationType } from '@eggjs/tegg/transaction';
+import assert from 'node:assert';
+import { MysqlDataSource } from '@eggjs/dal-runtime';
+
+export interface TransactionalParams {
+  propagation: PropagationType;
+  dataSourceGetter: () => MysqlDataSource;
+}
+
+@Advice({
+  accessLevel: AccessLevel.PUBLIC,
+  initType: ObjectInitType.SINGLETON,
+})
+export class TransactionalAOP implements IAdvice<EggProtoImplClass, TransactionalParams> {
+  public async around(ctx: AdviceContext<EggProtoImplClass, TransactionalParams>, next: () => Promise<any>): Promise<void> {
+    const { propagation, dataSourceGetter } = ctx.adviceParams!;
+    const dataSource = dataSourceGetter();
+    assert(propagation === PropagationType.REQUIRED, '事务注解目前只支持 REQUIRED 机制');
+    return await dataSource.beginTransactionScope(next);
+  }
+}

plugin/dal/package.json

@@ -55,6 +55,7 @@
   "devDependencies": {
     "@eggjs/tegg-config": "^3.37.3",
     "@eggjs/tegg-plugin": "^3.37.3",
+    "@eggjs/tegg-aop-plugin": "^3.37.3",
     "@types/mocha": "^10.0.1",
     "@types/node": "^20.2.4",
     "cross-env": "^7.0.3",

plugin/dal/test/fixtures/apps/dal-app/config/plugin.js

@@ -14,3 +14,8 @@ exports.teggConfig = {
   package: '@eggjs/tegg-config',
   enable: true,
 };
+
+exports.aopModule = {
+  enable: true,
+  package: '@eggjs/tegg-aop-plugin',
+};

plugin/dal/test/fixtures/apps/dal-app/modules/dal/Foo.ts

@@ -294,4 +294,84 @@ export class Foo {
     type: ColumnType.JSON,
   })
   jsonColumn: object;
+
+  static buildObj() {
+    const foo = new Foo();
+    foo.name = 'name';
+    foo.col1 = 'col1';
+    foo.bitColumn = Buffer.from([ 0, 0 ]);
+    foo.boolColumn = 0;
+    foo.tinyIntColumn = 0;
+    foo.smallIntColumn = 1;
+    foo.mediumIntColumn = 3;
+    foo.intColumn = 3;
+    foo.bigIntColumn = '00099';
+    foo.decimalColumn = '00002.33333';
+    foo.floatColumn = 2.3;
+    foo.doubleColumn = 2.3;
+    foo.dateColumn = new Date('2020-03-15T16:00:00.000Z');
+    foo.dateTimeColumn = new Date('2024-03-16T01:26:58.677Z');
+    foo.timestampColumn = new Date('2024-03-16T01:26:58.677Z');
+    foo.timeColumn = '838:59:50.123';
+    foo.yearColumn = 2024;
+    foo.varCharColumn = 'var_char';
+    foo.binaryColumn = Buffer.from('b');
+    foo.varBinaryColumn = Buffer.from('var_binary');
+    foo.tinyBlobColumn = Buffer.from('tiny_blob');
+    foo.tinyTextColumn = 'text';
+    foo.blobColumn = Buffer.from('blob');
+    foo.textColumn = 'text';
+    foo.mediumBlobColumn = Buffer.from('medium_blob');
+    foo.longBlobColumn = Buffer.from('long_blob');
+    foo.mediumTextColumn = 'medium_text';
+    foo.longTextColumn = 'long_text';
+    foo.enumColumn = 'A';
+    foo.setColumn = 'B';
+    foo.geometryColumn = { x: 10, y: 10 };
+    foo.pointColumn = { x: 10, y: 10 };
+    foo.lineStringColumn = [
+      { x: 15, y: 15 },
+      { x: 20, y: 20 },
+    ];
+    foo.polygonColumn = [
+      [
+        { x: 0, y: 0 }, { x: 10, y: 0 }, { x: 10, y: 10 }, { x: 0, y: 10 }, { x: 0, y: 0 },
+      ], [
+        { x: 5, y: 5 }, { x: 7, y: 5 }, { x: 7, y: 7 }, { x: 5, y: 7 }, { x: 5, y: 5 },
+      ],
+    ];
+    foo.multipointColumn = [
+      { x: 0, y: 0 }, { x: 20, y: 20 }, { x: 60, y: 60 },
+    ];
+    foo.multiLineStringColumn = [
+      [
+        { x: 10, y: 10 }, { x: 20, y: 20 },
+      ], [
+        { x: 15, y: 15 }, { x: 30, y: 15 },
+      ],
+    ];
+    foo.multiPolygonColumn = [
+      [
+        [
+          { x: 0, y: 0 }, { x: 10, y: 0 }, { x: 10, y: 10 }, { x: 0, y: 10 }, { x: 0, y: 0 },
+        ],
+      ],
+      [
+        [
+          { x: 5, y: 5 }, { x: 7, y: 5 }, { x: 7, y: 7 }, { x: 5, y: 7 }, { x: 5, y: 5 },
+        ],
+      ],
+    ];
+    foo.geometryCollectionColumn = [
+      { x: 10, y: 10 },
+      { x: 30, y: 30 },
+      [
+        { x: 15, y: 15 }, { x: 20, y: 20 },
+      ],
+    ];
+    foo.jsonColumn = {
+      hello: 'json',
+    };
+    return foo;
+  }
 }

plugin/dal/test/fixtures/apps/dal-app/modules/dal/FooService.ts

@@ -0,0 +1,33 @@
+import { AccessLevel, Inject, SingletonProto } from '@eggjs/tegg';
+import { Transactional } from '@eggjs/tegg/transaction';
+import FooDAO from './dal/dao/FooDAO';
+import { Foo } from './Foo';
+
+@SingletonProto({
+  accessLevel: AccessLevel.PUBLIC,
+})
+export class FooService {
+  @Inject()
+  private readonly fooDAO: FooDAO;
+
+  @Transactional()
+  async succeedTransaction() {
+    const foo = Foo.buildObj();
+    foo.name = 'insert_succeed_transaction_1';
+    const foo2 = Foo.buildObj();
+    foo2.name = 'insert_succeed_transaction_2';
+    await this.fooDAO.insert(foo);
+    await this.fooDAO.insert(foo2);
+  }
+
+  @Transactional()
+  async failedTransaction() {
+    const foo = Foo.buildObj();
+    foo.name = 'insert_failed_transaction_1';
+    const foo2 = Foo.buildObj();
+    foo2.name = 'insert_failed_transaction_2';
+    await this.fooDAO.insert(foo);
+    await this.fooDAO.insert(foo2);
+    throw new Error('mock error');
+  }
+}

plugin/dal/test/transaction.test.ts

@@ -0,0 +1,89 @@
+import assert from 'assert';
+import path from 'path';
+import mm, { MockApplication } from 'egg-mock';
+import FooDAO from './fixtures/apps/dal-app/modules/dal/dal/dao/FooDAO';
+import { FooService } from './fixtures/apps/dal-app/modules/dal/FooService';
+import { MysqlDataSourceManager } from '../lib/MysqlDataSourceManager';
+
+describe('plugin/dal/test/transaction.test.ts', () => {
+  let app: MockApplication;
+
+  afterEach(async () => {
+    mm.restore();
+  });
+
+  before(async () => {
+    mm(process.env, 'EGG_TYPESCRIPT', true);
+    mm(process, 'cwd', () => {
+      return path.join(__dirname, '../');
+    });
+    app = mm.app({
+      baseDir: path.join(__dirname, './fixtures/apps/dal-app'),
+      framework: require.resolve('egg'),
+    });
+    await app.ready();
+  });
+
+  afterEach(async () => {
+    const dataSource = MysqlDataSourceManager.instance.get('dal', 'foo')!;
+    await dataSource.query('delete from egg_foo;');
+  });
+
+  after(() => {
+    return app.close();
+  });
+
+  describe('succeed transaction', () => {
+    it('should commit', async () => {
+      await app.mockModuleContextScope(async () => {
+        const fooService = await app.getEggObject(FooService);
+        const fooDao = await app.getEggObject(FooDAO);
+        await fooService.succeedTransaction();
+        const foo1 = await fooDao.findByName('insert_succeed_transaction_1');
+        const foo2 = await fooDao.findByName('insert_succeed_transaction_2');
+        assert(foo1.length);
+        assert(foo2.length);
+      });
+    });
+  });
+
+  describe('failed transaction', () => {
+    it('should rollback', async () => {
+      await app.mockModuleContextScope(async () => {
+        const fooService = await app.getEggObject(FooService);
+        const fooDao = await app.getEggObject(FooDAO);
+        await assert.rejects(async () => {
+          await fooService.failedTransaction();
+        });
+        const foo1 = await fooDao.findByName('insert_failed_transaction_1');
+        const foo2 = await fooDao.findByName('insert_failed_transaction_2');
+        assert(!foo1.length);
+        assert(!foo2.length);
+      });
+    });
+  });
+
+  describe('transaction should be isolated', () => {
+    it('should rollback', async () => {
+      await app.mockModuleContextScope(async () => {
+        const fooService = await app.getEggObject(FooService);
+        const fooDao = await app.getEggObject(FooDAO);
+        const [ failedRes, succeedRes ] = await Promise.allSettled([
+          fooService.failedTransaction(),
+          fooService.succeedTransaction(),
+        ]);
+        assert.equal(failedRes.status, 'rejected');
+        assert.equal(succeedRes.status, 'fulfilled');
+        const foo1 = await fooDao.findByName('insert_failed_transaction_1');
+        const foo2 = await fooDao.findByName('insert_failed_transaction_2');
+        assert(!foo1.length);
+        assert(!foo2.length);
+
+        const foo3 = await fooDao.findByName('insert_succeed_transaction_1');
+        const foo4 = await fooDao.findByName('insert_succeed_transaction_2');
+        assert(foo3.length);
+        assert(foo4.length);
+      });
+    });
+  });
+});

standalone/standalone/src/Runner.ts

@@ -35,6 +35,7 @@ import { DalModuleLoadUnitHook } from '@eggjs/tegg-dal-plugin/lib/DalModuleLoadU
 import { MysqlDataSourceManager } from '@eggjs/tegg-dal-plugin/lib/MysqlDataSourceManager';
 import { SqlMapManager } from '@eggjs/tegg-dal-plugin/lib/SqlMapManager';
 import { TableModelManager } from '@eggjs/tegg-dal-plugin/lib/TableModelManager';
+import { TransactionPrototypeHook } from '@eggjs/tegg-dal-plugin/lib/TransactionPrototypeHook';
 
 export interface ModuleDependency extends ReadModuleReferenceOptions {
   baseDir: string;
@@ -64,6 +65,7 @@ export class Runner {
   private loadUnitMultiInstanceProtoHook: LoadUnitMultiInstanceProtoHook;
   private dalTableEggPrototypeHook: DalTableEggPrototypeHook;
   private dalModuleLoadUnitHook: DalModuleLoadUnitHook;
+  private transactionPrototypeHook: TransactionPrototypeHook;
 
   private readonly loadUnitInnerClassHook: LoadUnitInnerClassHook;
   private readonly crosscutAdviceFactory: CrosscutAdviceFactory;
@@ -156,9 +158,11 @@ export class Runner {
 
     this.dalModuleLoadUnitHook = new DalModuleLoadUnitHook(this.env ?? '', this.moduleConfigs);
     const loggerInnerObject = this.innerObjects.logger && this.innerObjects.logger[0];
-    const logger = loggerInnerObject?.obj || console;
-    this.dalTableEggPrototypeHook = new DalTableEggPrototypeHook(logger as Logger);
+    const logger = (loggerInnerObject?.obj || console) as Logger;
+    this.dalTableEggPrototypeHook = new DalTableEggPrototypeHook(logger);
+    this.transactionPrototypeHook = new TransactionPrototypeHook(this.moduleConfigs, logger);
     EggPrototypeLifecycleUtil.registerLifecycle(this.dalTableEggPrototypeHook);
+    EggPrototypeLifecycleUtil.registerLifecycle(this.transactionPrototypeHook);
     LoadUnitLifecycleUtil.registerLifecycle(this.dalModuleLoadUnitHook);
   }
 
@@ -257,6 +261,9 @@ export class Runner {
     if (this.dalModuleLoadUnitHook) {
       LoadUnitLifecycleUtil.deleteLifecycle(this.dalModuleLoadUnitHook);
     }
+    if (this.transactionPrototypeHook) {
+      EggPrototypeLifecycleUtil.deleteLifecycle(this.transactionPrototypeHook);
+    }
     MysqlDataSourceManager.instance.clear();
     SqlMapManager.instance.clear();
     TableModelManager.instance.clear();