feat: add ability to specify knex transaction config

packages/entity-database-adapter-knex/src/PostgresEntityQueryContextProvider.ts

@@ -1,4 +1,8 @@
-import { EntityQueryContextProvider } from '@expo/entity';
+import {
+  EntityQueryContextProvider,
+  TransactionConfig,
+  TransactionIsolationLevel,
+} from '@expo/entity';
 import { Knex } from 'knex';
 
 /**
@@ -13,15 +17,44 @@ export default class PostgresEntityQueryContextProvider extends EntityQueryConte
     return this.knexInstance;
   }
 
-  protected createTransactionRunner<T>(): (
-    transactionScope: (trx: any) => Promise<T>
-  ) => Promise<T> {
-    return (transactionScope) => this.knexInstance.transaction(transactionScope);
+  protected createTransactionRunner<T>(
+    transactionConfig?: TransactionConfig
+  ): (transactionScope: (trx: any) => Promise<T>) => Promise<T> {
+    return (transactionScope) =>
+      this.knexInstance.transaction(
+        transactionScope,
+        transactionConfig
+          ? PostgresEntityQueryContextProvider.convertTransactionConfig(transactionConfig)
+          : undefined
+      );
   }
 
   protected createNestedTransactionRunner<T>(
     outerQueryInterface: any
   ): (transactionScope: (queryInterface: any) => Promise<T>) => Promise<T> {
     return (transactionScope) => (outerQueryInterface as Knex).transaction(transactionScope);
   }
+
+  private static convertTransactionConfig(
+    transactionConfig: TransactionConfig
+  ): Knex.TransactionConfig {
+    const convertIsolationLevel = (
+      isolationLevel: TransactionIsolationLevel
+    ): Knex.IsolationLevels => {
+      switch (isolationLevel) {
+        case TransactionIsolationLevel.READ_COMMITTED:
+          return 'read committed';
+        case TransactionIsolationLevel.REPEATABLE_READ:
+          return 'repeatable read';
+        case TransactionIsolationLevel.SERIALIZABLE:
+          return 'serializable';
+      }
+    };
+
+    return {
+      ...(transactionConfig.isolationLevel
+        ? { isolationLevel: convertIsolationLevel(transactionConfig.isolationLevel) }
+        : {}),
+    };
+  }
 }

packages/entity-database-adapter-knex/src/__integration-tests__/PostgresEntityIntegration-test.ts

@@ -3,6 +3,7 @@ import {
   createUnitTestEntityCompanionProvider,
   enforceResultsAsync,
   ViewerContext,
+  TransactionIsolationLevel,
 } from '@expo/entity';
 import { enforceAsyncResult } from '@expo/results';
 import { knex, Knex } from 'knex';
@@ -111,6 +112,44 @@ describe('postgres entity integration', () => {
     expect(entities).toHaveLength(1);
   });
 
+  it('passes transaction config into transactions', async () => {
+    const vc1 = new ViewerContext(createKnexIntegrationTestEntityCompanionProvider(knexInstance));
+
+    const firstEntity = await enforceAsyncResult(
+      PostgresTestEntity.creator(vc1).setField('name', 'hello').createAsync()
+    );
+
+    const loadAndUpdateAsync = async (newName: string): Promise<{ error?: Error }> => {
+      try {
+        await vc1.runInTransactionForDatabaseAdaptorFlavorAsync(
+          'postgres',
+          async (queryContext) => {
+            const entity = await PostgresTestEntity.loader(vc1, queryContext)
+              .enforcing()
+              .loadByIDAsync(firstEntity.getID());
+            await PostgresTestEntity.updater(entity, queryContext)
+              .setField('name', newName)
+              .enforceUpdateAsync();
+          },
+          { isolationLevel: TransactionIsolationLevel.SERIALIZABLE }
+        );
+        return {};
+      } catch (e) {
+        return { error: e as Error };
+      }
+    };
+
+    // do some parallel updates to trigger serializable error in at least some of them
+    const results = await Promise.all([
+      loadAndUpdateAsync('hello2'),
+      loadAndUpdateAsync('hello3'),
+      loadAndUpdateAsync('hello4'),
+      loadAndUpdateAsync('hello5'),
+    ]);
+
+    expect(results.filter((r) => (r.error as any)?.cause?.code === '40001').length > 0).toBe(true);
+  });
+
   describe('JSON fields', () => {
     it('supports both types of array fields', async () => {
       const vc1 = new ViewerContext(createKnexIntegrationTestEntityCompanionProvider(knexInstance));

packages/entity-example/src/adapters/InMemoryQueryContextProvider.ts

@@ -1,13 +1,13 @@
-import { EntityQueryContextProvider } from '@expo/entity';
+import { EntityQueryContextProvider, TransactionConfig } from '@expo/entity';
 
 export default class InMemoryQueryContextProvider extends EntityQueryContextProvider {
   protected getQueryInterface(): any {
     return {};
   }
 
-  protected createTransactionRunner<T>(): (
-    transactionScope: (queryInterface: any) => Promise<T>
-  ) => Promise<T> {
+  protected createTransactionRunner<T>(
+    _transactionConfig?: TransactionConfig
+  ): (transactionScope: (queryInterface: any) => Promise<T>) => Promise<T> {
     return (transactionScope) => Promise.resolve(transactionScope({}));
   }
 

packages/entity/src/EntityQueryContext.ts

@@ -8,6 +8,16 @@ export type PreCommitCallback = (
   ...args: any
 ) => Promise<any>;
 
+export enum TransactionIsolationLevel {
+  READ_COMMITTED = 'READ_COMMITTED',
+  REPEATABLE_READ = 'REPEATABLE_READ',
+  SERIALIZABLE = 'SERIALIZABLE',
+}
+
+export type TransactionConfig = {
+  isolationLevel?: TransactionIsolationLevel;
+};
+
 /**
  * Entity framework representation of transactional and non-transactional database
  * query execution units.
@@ -25,7 +35,8 @@ export abstract class EntityQueryContext {
   }
 
   abstract runInTransactionIfNotInTransactionAsync<T>(
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>
+    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>,
+    transactionConfig?: TransactionConfig
   ): Promise<T>;
 }
 
@@ -48,9 +59,13 @@ export class EntityNonTransactionalQueryContext extends EntityQueryContext {
   }
 
   async runInTransactionIfNotInTransactionAsync<T>(
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>
+    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>,
+    transactionConfig?: TransactionConfig
   ): Promise<T> {
-    return await this.entityQueryContextProvider.runInTransactionAsync(transactionScope);
+    return await this.entityQueryContextProvider.runInTransactionAsync(
+      transactionScope,
+      transactionConfig
+    );
   }
 }
 
@@ -132,8 +147,13 @@ export class EntityTransactionalQueryContext extends EntityQueryContext {
   }
 
   async runInTransactionIfNotInTransactionAsync<T>(
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>
+    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>,
+    transactionConfig?: TransactionConfig
   ): Promise<T> {
+    assert(
+      transactionConfig === undefined,
+      'Should not pass transactionConfig to a nested transaction'
+    );
     return await transactionScope(this);
   }
 

packages/entity/src/EntityQueryContextProvider.ts

@@ -2,6 +2,7 @@ import {
   EntityTransactionalQueryContext,
   EntityNonTransactionalQueryContext,
   EntityNestedTransactionalQueryContext,
+  TransactionConfig,
 } from './EntityQueryContext';
 
 /**
@@ -23,9 +24,9 @@ export default abstract class EntityQueryContextProvider {
   /**
    * Vend a transaction runner for use in runInTransactionAsync.
    */
-  protected abstract createTransactionRunner<T>(): (
-    transactionScope: (queryInterface: any) => Promise<T>
-  ) => Promise<T>;
+  protected abstract createTransactionRunner<T>(
+    transactionConfig?: TransactionConfig
+  ): (transactionScope: (queryInterface: any) => Promise<T>) => Promise<T>;
 
   protected abstract createNestedTransactionRunner<T>(
     outerQueryInterface: any
@@ -36,11 +37,12 @@ export default abstract class EntityQueryContextProvider {
    * @param transactionScope - async callback to execute within the transaction
    */
   async runInTransactionAsync<T>(
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>
+    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>,
+    transactionConfig?: TransactionConfig
   ): Promise<T> {
     const [returnedValue, queryContext] = await this.createTransactionRunner<
       [T, EntityTransactionalQueryContext]
-    >()(async (queryInterface) => {
+    >(transactionConfig)(async (queryInterface) => {
       const queryContext = new EntityTransactionalQueryContext(queryInterface, this);
       const result = await transactionScope(queryContext);
       await queryContext.runPreCommitCallbacksAsync();

packages/entity/src/ViewerContext.ts

@@ -1,7 +1,11 @@
 import { IEntityClass } from './Entity';
 import EntityCompanionProvider, { DatabaseAdapterFlavor } from './EntityCompanionProvider';
 import EntityPrivacyPolicy from './EntityPrivacyPolicy';
-import { EntityQueryContext, EntityTransactionalQueryContext } from './EntityQueryContext';
+import {
+  EntityQueryContext,
+  EntityTransactionalQueryContext,
+  TransactionConfig,
+} from './EntityQueryContext';
 import ReadonlyEntity from './ReadonlyEntity';
 import ViewerScopedEntityCompanion from './ViewerScopedEntityCompanion';
 import ViewerScopedEntityCompanionProvider from './ViewerScopedEntityCompanionProvider';
@@ -82,11 +86,12 @@ export default class ViewerContext {
    */
   async runInTransactionForDatabaseAdaptorFlavorAsync<TResult>(
     databaseAdaptorFlavor: DatabaseAdapterFlavor,
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<TResult>
+    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<TResult>,
+    transactionConfig?: TransactionConfig
   ): Promise<TResult> {
     return await this.entityCompanionProvider
       .getQueryContextProviderForDatabaseAdaptorFlavor(databaseAdaptorFlavor)
       .getQueryContext()
-      .runInTransactionIfNotInTransactionAsync(transactionScope);
+      .runInTransactionIfNotInTransactionAsync(transactionScope, transactionConfig);
   }
 }

packages/entity/src/__tests__/EntityQueryContext-test.ts

@@ -1,6 +1,6 @@
 import invariant from 'invariant';
 
-import { EntityQueryContext } from '../EntityQueryContext';
+import { EntityQueryContext, TransactionIsolationLevel } from '../EntityQueryContext';
 import ViewerContext from '../ViewerContext';
 import { createUnitTestEntityCompanionProvider } from '../utils/testing/createUnitTestEntityCompanionProvider';
 
@@ -129,4 +129,26 @@ describe(EntityQueryContext, () => {
       expect(postCommitInvalidationCallback).toHaveBeenCalledTimes(2);
     });
   });
+
+  describe('transaction config', () => {
+    it('passes it into the provider', async () => {
+      const companionProvider = createUnitTestEntityCompanionProvider();
+      const viewerContext = new ViewerContext(companionProvider);
+
+      const queryContextProvider =
+        companionProvider.getQueryContextProviderForDatabaseAdaptorFlavor('postgres');
+      const queryContextProviderSpy = jest.spyOn(queryContextProvider, 'runInTransactionAsync');
+
+      const transactionScopeFn = async (): Promise<any> => {};
+      const transactionConfig = { isolationLevel: TransactionIsolationLevel.SERIALIZABLE };
+
+      await viewerContext.runInTransactionForDatabaseAdaptorFlavorAsync(
+        'postgres',
+        transactionScopeFn,
+        transactionConfig
+      );
+
+      expect(queryContextProviderSpy).toHaveBeenCalledWith(transactionScopeFn, transactionConfig);
+    });
+  });
 });

packages/entity/src/internal/__tests__/EntityDataManager-test.ts

@@ -385,7 +385,8 @@ describe(EntityDataManager, () => {
         return await entityDataManager.loadManyByFieldEqualingAsync(queryContext, 'customIdField', [
           '1',
         ]);
-      }
+      },
+      {}
     );
 
     expect(entityDatas.get('1')).toHaveLength(1);

packages/entity/src/utils/testing/StubQueryContextProvider.ts

@@ -1,13 +1,14 @@
+import { TransactionConfig } from '../../EntityQueryContext';
 import EntityQueryContextProvider from '../../EntityQueryContextProvider';
 
 export class StubQueryContextProvider extends EntityQueryContextProvider {
   protected getQueryInterface(): any {
     return {};
   }
 
-  protected createTransactionRunner<T>(): (
-    transactionScope: (queryInterface: any) => Promise<T>
-  ) => Promise<T> {
+  protected createTransactionRunner<T>(
+    _transactionConfig?: TransactionConfig
+  ): (transactionScope: (queryInterface: any) => Promise<T>) => Promise<T> {
     return (transactionScope) => Promise.resolve(transactionScope({}));
   }