Fixes test but ideally want to make transactional

Signed-off-by: John Gomersall <[email protected]>

src/domain/services/import.service.spec.ts

@@ -1,6 +1,6 @@
 import { DomainModule } from '../domain.module';
 import { ImportService } from './import.service';
-import { EntityManager } from '@mikro-orm/core';
+import { EntityManager } from '@mikro-orm/postgresql';
 import { Product } from '../entities/product';
 import { ProductIngredientsTag } from '../entities/product-tags';
 import { createTestingModule, randomCode } from '../../../test/test.helper';
@@ -351,7 +351,7 @@ describe('ProductTag', () => {
 });
 
 describe('importWithFilter', () => {
-  it.only('should not get an error with concurrent imports', async () => {
+  it('should not get an error with concurrent imports', async () => {
     await createTestingModule([DomainModule], async (app) => {
       const importService = app.get(ImportService);
 

src/domain/services/import.service.ts

@@ -131,7 +131,7 @@ export class ImportService {
     await client.close();
 
     // Tags are popualted using raw SQL from the data field
-    await this.updateTags(updateId, fullImport);
+    await this.updateTags(updateId, fullImport, source);
 
     // If doing a full import delete all products that weren't updated
     if (fullImport) {
@@ -236,14 +236,20 @@ export class ImportService {
    * SQL is then run to insert this into the individual tag tables.
    * This was found to be quicker than using ORM functionality
    */
-  async updateTags(updateId: string, fullImport = false) {
+  async updateTags(
+    updateId: string,
+    fullImport = false,
+    source = ProductSource.FULL_LOAD,
+  ) {
+    const autoCommit = source !== ProductSource.EVENT;
+
     this.logger.debug(`Updating tags for updateId: ${updateId}`);
 
     const connection = this.em.getConnection();
 
     // Fix ingredients
     let logText = `Updated ingredients`;
-    await connection.execute('begin');
+    if (autoCommit) await connection.execute('begin');
     const deleted = await connection.execute(
       `delete from product_ingredient 
     where product_id in (select id from product 
@@ -329,15 +335,15 @@ export class ImportService {
       affectedRows = results['affectedRows'];
       logText += ` > ${affectedRows}`;
     }
-    await connection.execute('commit');
+    if (autoCommit) await connection.execute('commit');
     this.logger.debug(logText + ' rows');
 
     for (const [tag, entity] of Object.entries(ProductTagMap.MAPPED_TAGS)) {
       let logText = `Updated ${tag}`;
       // Get the underlying table name for the entity
       const tableName = this.em.getMetadata(entity).tableName;
 
-      await connection.execute('begin');
+      if (autoCommit) await connection.execute('begin');
 
       // Delete existing tags for products that were imorted on this run
       const deleted = await connection.execute(
@@ -360,7 +366,7 @@ export class ImportService {
       );
 
       // Commit after each tag to minimise server snapshot size
-      await connection.execute('commit');
+      if (autoCommit) await connection.execute('commit');
 
       // If this is a full load we can flag the tag as now available for query
       if (fullImport) {