[TECH] Eviter de créer des transactions nestées lorsqu'on utilise la fonction batchInsert de knex dans le cadre d'une DomainTransaction

 #11435

Author: pix-service-auto-merge

Diff for: api/lib/infrastructure/repositories/target-profile-share-repository.js

@@ -1,8 +1,11 @@
+import { knex } from '../../../db/knex-database-connection.js';
 import { DomainTransaction } from '../../../src/shared/domain/DomainTransaction.js';
 
 const batchAddTargetProfilesToOrganization = async function (organizationTargetProfiles) {
   const knexConn = DomainTransaction.getConnection();
-  await knexConn.batchInsert('target-profile-shares', organizationTargetProfiles);
+  await knex
+    .batchInsert('target-profile-shares', organizationTargetProfiles)
+    .transacting(knexConn.isTransaction ? knexConn : null);
 };
 
 export { batchAddTargetProfilesToOrganization };

Diff for: api/scripts/certification/import-certification-cpf-cities.js

@@ -412,7 +412,7 @@ async function main(filePath) {
     logger.info('Inserting cities in database... ');
     trx = await knex.transaction();
     await trx('certification-cpf-cities').del();
-    const batchInfo = await trx.batchInsert('certification-cpf-cities', cities);
+    const batchInfo = await knex.batchInsert('certification-cpf-cities', cities).transacting(trx);
     const insertedLines = _getInsertedLineNumber(batchInfo);
     logger.info('✅ ');
     await trx.commit();

Diff for: api/scripts/certification/import-certification-cpf-countries.js

@@ -97,7 +97,7 @@ async function main(filePath) {
     console.log('Emptying existing countries in database... ');
     await trx('certification-cpf-countries').del();
     console.log('Inserting countries in database... ');
-    await trx.batchInsert('certification-cpf-countries', countries);
+    await knex.batchInsert('certification-cpf-countries', countries).transacting(trx);
     await trx.commit();
     console.log('ok');
 

Diff for: api/scripts/certification/next-gen/import-complementary-alone-feature-pilot-certification-centers-from-csv.js

@@ -92,7 +92,9 @@ async function main(filePath) {
       .leftJoin('features', 'features.id', 'certification-center-features.featureId')
       .where('features.key', CERTIFICATION_FEATURES.CAN_REGISTER_FOR_A_COMPLEMENTARY_CERTIFICATION_ALONE.key)
       .del();
-    const batchInfo = await trx.batchInsert('certification-center-features', certificationCentersPilotsList);
+    const batchInfo = await knex
+      .batchInsert('certification-center-features', certificationCentersPilotsList)
+      .transacting(trx);
     const insertedLines = _getInsertedLineNumber(batchInfo);
     logger.info('✅ ');
     await trx.commit();

Diff for: api/scripts/data-generation/generate-campaign-with-participants.js

@@ -268,7 +268,7 @@ async function _createUsers({ count, uniqId, trx }) {
     });
   }
   const chunkSize = _getChunkSize(userData[0]);
-  const users = await trx.batchInsert('users', userData.flat(), chunkSize).returning('id');
+  const users = await knex.batchInsert('users', userData.flat(), chunkSize).transacting(trx).returning('id');
   return users.map((user) => user.id);
 }
 
@@ -296,8 +296,9 @@ async function _createOrganizationLearners({ userIds, organizationId, uniqId, tr
     organizationLearnerData.push(organizationLearnerSpecificBuilder({ userId, organizationId, identifier }));
   }
   const chunkSize = _getChunkSize(organizationLearnerData[0]);
-  return trx
+  return knex
     .batchInsert('organization-learners', organizationLearnerData.flat(), chunkSize)
+    .transacting(trx)
     .returning(['id', 'userId']);
 }
 
@@ -355,7 +356,7 @@ async function _createAssessments({ userAndCampaignParticipationIds, trx }) {
     });
   }
   const chunkSize = _getChunkSize(assessmentData[0]);
-  return trx.batchInsert('assessments', assessmentData.flat(), chunkSize).returning(['id', 'userId']);
+  return knex.batchInsert('assessments', assessmentData.flat(), chunkSize).transacting(trx).returning(['id', 'userId']);
 }
 
 async function _createCampaignParticipations({ campaignId, trx, organizationLearnerAndUserIds }) {
@@ -377,7 +378,10 @@ async function _createCampaignParticipations({ campaignId, trx, organizationLear
     });
   }
   const chunkSize = _getChunkSize(participationData[0]);
-  return trx.batchInsert('campaign-participations', participationData.flat(), chunkSize).returning(['id', 'userId']);
+  return knex
+    .batchInsert('campaign-participations', participationData.flat(), chunkSize)
+    .transacting(trx)
+    .returning(['id', 'userId']);
 }
 
 async function _createAnswersAndKnowledgeElements({ campaignId, userAndAssessmentIds, trx }) {
@@ -391,8 +395,9 @@ async function _createAnswersAndKnowledgeElements({ campaignId, userAndAssessmen
     });
   }
   const chunkSize = _getChunkSize(answerData[0]);
-  const answerRecordedData = await trx
+  const answerRecordedData = await knex
     .batchInsert('answers', answerData.flat(), chunkSize)
+    .transacting(trx)
     .returning(['id', 'assessmentId']);
   _log('\tOK');
 
@@ -467,7 +472,7 @@ async function _createBadgeAcquisitions({ targetProfile, userAndCampaignParticip
     }
   }
   const chunkSize = _getChunkSize(badgeAcquisitionData[0]);
-  await trx.batchInsert('badge-acquisitions', badgeAcquisitionData.flat(), chunkSize);
+  await knex.batchInsert('badge-acquisitions', badgeAcquisitionData.flat(), chunkSize).transacting(trx);
   _log(`\t${badgeAcquisitionData.flat().length} acquisitions de badge créées`);
 }
 

Diff for: api/scripts/prod/target-profile-migrations/common.js

@@ -1,5 +1,6 @@
 import _ from 'lodash';
 
+import { knex } from '../../../db/knex-database-connection.js';
 import * as skillRepository from '../../../src/shared/infrastructure/repositories/skill-repository.js';
 import * as tubeRepository from '../../../src/shared/infrastructure/repositories/tube-repository.js';
 
@@ -37,7 +38,7 @@ const autoMigrateTargetProfile = async function (id, trx) {
     return { ...tube, targetProfileId: id };
   });
   await trx('target-profiles').update({ migration_status: 'AUTO' }).where({ id });
-  await trx.batchInsert('target-profile_tubes', completeTubes);
+  await knex.batchInsert('target-profile_tubes', completeTubes).transacting(trx);
 };
 
 export { autoMigrateTargetProfile };

Diff for: api/src/certification/enrolment/infrastructure/repositories/sco-certification-candidate-repository.js

@@ -29,7 +29,7 @@ const addNonEnrolledCandidatesToSession = async function ({ sessionId, scoCertif
         allSubscriptionsDTO.push(subscriptionDTO);
       }
     }
-    await trx.batchInsert('certification-subscriptions', allSubscriptionsDTO);
+    await knex.batchInsert('certification-subscriptions', allSubscriptionsDTO).transacting(trx);
   });
 };
 

Diff for: api/src/organizational-entities/infrastructure/repositories/data-protection-officer.repository.js

@@ -1,11 +1,14 @@
+import { knex } from '../../../../db/knex-database-connection.js';
 import { DomainTransaction } from '../../../shared/domain/DomainTransaction.js';
 import { DataProtectionOfficer } from '../../domain/models/DataProtectionOfficer.js';
 
 const DATA_PROTECTION_OFFICERS_TABLE_NAME = 'data-protection-officers';
 
 async function batchAddDataProtectionOfficerToOrganization(organizationDataProtectionOfficer) {
   const knexConn = DomainTransaction.getConnection();
-  await knexConn.batchInsert(DATA_PROTECTION_OFFICERS_TABLE_NAME, organizationDataProtectionOfficer);
+  await knex
+    .batchInsert(DATA_PROTECTION_OFFICERS_TABLE_NAME, organizationDataProtectionOfficer)
+    .transacting(knexConn.isTransaction ? knexConn : null);
 }
 
 async function get({ organizationId = null, certificationCenterId = null }) {

Diff for: api/src/organizational-entities/infrastructure/repositories/organization-tag.repository.js

@@ -28,7 +28,7 @@ const create = async function (organizationTag) {
 const batchCreate = async function (organizationsTags) {
   const knexConn = DomainTransaction.getConnection();
 
-  return knexConn.batchInsert('organization-tags', organizationsTags);
+  return knex.batchInsert('organization-tags', organizationsTags).transacting(knexConn.isTransaction ? knexConn : null);
 };
 
 const isExistingByOrganizationIdAndTagId = async function ({ organizationId, tagId }) {

Diff for: api/src/prescription/campaign/infrastructure/repositories/campaign-administration-repository.js

@@ -113,7 +113,7 @@ const save = async function (campaigns, dependencies = { skillRepository }) {
             ...rightLevelSkills.map((skill) => ({ skillId: skill.id, campaignId: latestCreatedCampaign.id })),
           );
         }
-        await trx.batchInsert('campaign_skills', skillData);
+        await knex.batchInsert('campaign_skills', skillData).transacting(trx);
       }
     }
     await trx.commit();

Diff for: api/src/prescription/target-profile/infrastructure/repositories/target-profile-administration-repository.js

@@ -85,7 +85,7 @@ const create = async function ({ targetProfileForCreation }) {
     tubeId: tube.id,
     level: tube.level,
   }));
-  await knexConn.batchInsert('target-profile_tubes', tubesData);
+  await knex.batchInsert('target-profile_tubes', tubesData).transacting(knexConn.isTransaction ? knexConn : null);
 
   return targetProfileId;
 };

Diff for: api/src/shared/infrastructure/repositories/jobs/job-repository.js

@@ -1,5 +1,6 @@
 import Joi from 'joi';
 
+import { knex } from '../../../../../db/knex-database-connection.js';
 import { DomainTransaction } from '../../../domain/DomainTransaction.js';
 import { EntityValidationError } from '../../../domain/errors.js';
 
@@ -59,7 +60,7 @@ export class JobRepository {
   async #send(jobs) {
     const knexConn = DomainTransaction.getConnection();
 
-    const results = await knexConn.batchInsert('pgboss.job', jobs);
+    const results = await knex.batchInsert('pgboss.job', jobs).transacting(knexConn.isTransaction ? knexConn : null);
 
     const rowCount = results.reduce((total, batchResult) => total + (batchResult.rowCount || 0), 0);
 

Diff for: api/src/shared/infrastructure/repositories/knowledge-element-repository.js

@@ -44,7 +44,10 @@ const findUniqByUserIds = function (userIds) {
 const batchSave = async function ({ knowledgeElements }) {
   const knexConn = DomainTransaction.getConnection();
   const knowledgeElementsToSave = knowledgeElements.map((ke) => _.omit(ke, ['id', 'createdAt']));
-  const savedKnowledgeElements = await knexConn.batchInsert(tableName, knowledgeElementsToSave).returning('*');
+  const savedKnowledgeElements = await knex
+    .batchInsert(tableName, knowledgeElementsToSave)
+    .transacting(knexConn.isTransaction ? knexConn : null)
+    .returning('*');
   return savedKnowledgeElements.map((ke) => new KnowledgeElement(ke));
 };
 

Diff for: api/tests/evaluation/unit/infrastructure/repositories/answer-job-repository_test.js

@@ -1,11 +1,14 @@
 import { AnswerJobRepository } from '../../../../../src/evaluation/infrastructure/repositories/answer-job-repository.js';
 import { config } from '../../../../../src/shared/config.js';
 import { DomainTransaction } from '../../../../../src/shared/domain/DomainTransaction.js';
-import { expect, sinon } from '../../../../test-helper.js';
+import { expect, knex, sinon } from '../../../../test-helper.js';
 
 describe('Evaluation | Unit | Infrastructure | Repositories | AnswerJobRepository', function () {
   beforeEach(function () {
     sinon.stub(config, 'featureToggles');
+    sinon.stub(knex, 'batchInsert').callsFake(() => ({
+      transacting: sinon.stub().resolves([{ rowCount: 1 }]),
+    }));
     config.featureToggles.isQuestEnabled = true;
     config.featureToggles.isAsyncQuestRewardingCalculationEnabled = true;
   });

Diff for: api/tests/shared/integration/infrastructure/repositories/jobs/job-repository_test.js

@@ -1,11 +1,12 @@
+import { DomainTransaction } from '../../../../../../src/shared/domain/DomainTransaction.js';
 import { EntityValidationError } from '../../../../../../src/shared/domain/errors.js';
 import {
   JobExpireIn,
   JobPriority,
   JobRepository,
   JobRetry,
 } from '../../../../../../src/shared/infrastructure/repositories/jobs/job-repository.js';
-import { catchErrSync, expect } from '../../../../../test-helper.js';
+import { catchErr, catchErrSync, expect, knex } from '../../../../../test-helper.js';
 
 describe('Integration | Infrastructure | Repositories | Jobs | job-repository', function () {
   it('create one job db with given config', async function () {
@@ -63,6 +64,59 @@ describe('Integration | Infrastructure | Repositories | Jobs | job-repository',
     expect(jobsInserted.rowCount).to.equal(2);
   });
 
+  context('transaction', function () {
+    context('when no transaction ongoing', function () {
+      it("should not insert any jobs if one of them is invalid and can't be inserted", async function () {
+        // given
+        const name = 'JobTest';
+        // Knex doc : default chunk for batchInsert is 1000
+        const defaultChunkValidJobs = [...Array(1000).keys()].map((i) => ({ jobParam: i }));
+        const invalidJob = '>';
+        const priority = JobPriority.HIGH;
+        const job = new JobRepository({ name, priority });
+
+        // when
+        const expectedError = await catchErr(job.performAsync, job)(...defaultChunkValidJobs, invalidJob);
+
+        // then
+        expect(expectedError.detail).to.equal('Token ">" is invalid.');
+        const { count } = await knex('pgboss.job').count('id').first();
+        expect(count).to.equal(0);
+      });
+    });
+
+    context('when a transaction ongoing in DomainTransaction', function () {
+      it('should use the same existing transaction', async function () {
+        // given
+        const name = 'JobTest';
+        const jobs = [{ jobParam: 1 }, { jobParam: 2 }];
+        const priority = JobPriority.HIGH;
+        const job = new JobRepository({ name, priority });
+
+        // when
+        let knexConn;
+        const callback = async () => {
+          knexConn = DomainTransaction.getConnection();
+          await knexConn('features').insert({ key: 'someRandomFeature' });
+          await job.performAsync(...jobs);
+          const { count: countFeaturesBefore } = await knexConn('features').count('id').first();
+          expect(countFeaturesBefore).to.equal(1);
+          const { count: countJobsBefore } = await knexConn('pgboss.job').count('id').first();
+          expect(countJobsBefore).to.equal(2);
+          throw new Error('I want to rollback');
+        };
+        const expectedError = await catchErr(DomainTransaction.execute)(callback);
+
+        // then
+        expect(expectedError.message).to.equal('I want to rollback');
+        const { count: countFeaturesAfter } = await knex('features').count('id').first();
+        expect(countFeaturesAfter).to.equal(0);
+        const { count: countJobsAfter } = await knex('pgboss.job').count('id').first();
+        expect(countJobsAfter).to.equal(0);
+      });
+    });
+  });
+
   describe('JobExpireIn', function () {
     it('reject unexpected expiredIn value', async function () {
       // given