feat: consolidation des effectifs autour du RNCP/CFD

mission-apprentissage · Dec 11, 2024 · 65441c1 · 65441c1
1 parent aedf36d
commit 65441c1
Show file tree

Hide file tree

Showing 6 changed files with 189 additions and 32 deletions.
diff --git a/server/src/common/actions/engine/engine.actions.ts b/server/src/common/actions/engine/engine.actions.ts
@@ -1,6 +1,6 @@
 import { isEqual } from "date-fns";
-import { cloneDeep, get } from "lodash-es";
-import { Collection, WithoutId } from "mongodb";
+import { cloneDeep } from "lodash-es";
+import { Collection } from "mongodb";
 import { IEffectif } from "shared/models/data/effectifs.model";
 import { IEffectifDECA } from "shared/models/data/effectifsDECA.model";
 import { IEffectifQueue } from "shared/models/data/effectifsQueue.model";
@@ -91,19 +91,37 @@ export const completeEffectifAddress = async <T extends { apprenant: Partial<IEf
   return effectifDataWithAddress;
 };
 
-/**
- * Fonction de vérification de la présence d'un effectif via la clé d'unicité
- * id_erp_apprenant : identifiant unique du jeune dans le CFA
- * organisme_id : identifiant de l'organisme de formation en apprentissage
- * formation.cfd : Code formation diplôme de la formation suivie par le jeune
- * annee_scolaire : Année scolaire dans laquelle se trouve le jeune pour cette formation dans cet établissement
- */
-export const checkIfEffectifExists = async (effectif: IEffectif | WithoutId<IEffectifDECA>, db: Collection<any>) => {
-  const queryKeys = ["id_erp_apprenant", "organisme_id", "formation.cfd", "annee_scolaire"];
-  // Recherche de l'effectif via sa clé d'unicité
-  const query = queryKeys.reduce((acc, item) => ({ ...acc, [item]: get(effectif, item) }), {});
+export const checkIfEffectifExists = async <E extends IEffectif | IEffectifDECA>(
+  effectif: Pick<E, "id_erp_apprenant" | "organisme_id" | "annee_scolaire" | "formation">,
+  db: Collection<any>
+): Promise<E | null> => {
+  const effectifs = await db
+    .find({
+      id_erp_apprenant: effectif.id_erp_apprenant,
+      organisme_id: effectif.organisme_id,
+      annee_scolaire: effectif.annee_scolaire,
+    })
+    .toArray();
+
+  const newCfd = effectif.formation?.cfd ?? null;
+  const newRncp = effectif.formation;
+
+  const macthingEffectifs = effectifs.filter((eff) => {
+    const currentCfd = eff.formation?.cfd ?? null;
+    const currentRncp = eff.formation?.rncp ?? null;
+
+    // Si le CFD est null, on considère qu'on met à jour l'effectif
+    const isSameCfd = currentCfd === newCfd || currentCfd === null || newCfd === null;
+    const isSameRncp = currentRncp === newRncp || currentRncp === null || newRncp === null;
+
+    return isSameCfd && isSameRncp;
+  });
+
+  if (macthingEffectifs.length === 0) {
+    return null;
+  }
 
-  return await db.findOne(query);
+  return macthingEffectifs[0];
 };
 
 /**

diff --git a/server/src/db/migrations/20241211154816-deduplication-effectifs-auto.ts b/server/src/db/migrations/20241211154816-deduplication-effectifs-auto.ts
@@ -0,0 +1,123 @@
+import type { ObjectId } from "mongodb";
+import { MOTIF_SUPPRESSION } from "shared/constants";
+
+import { softDeleteEffectif } from "@/common/actions/effectifs.actions";
+import { effectifsDb, effectifsDECADb } from "@/common/model/collections";
+
+export const up = async () => {
+  const effectifDuplicats = await effectifsDb()
+    .aggregate<{
+      _id: {
+        annee_scolaire: string;
+        cfd: string;
+        rncp: string;
+        id_erp_apprenant: string;
+        organisme_id: string;
+      };
+      count: number;
+      docs: {
+        id: ObjectId;
+        updated_at: Date;
+      }[];
+    }>([
+      {
+        $group: {
+          _id: {
+            annee_scolaire: "$annee_scolaire",
+            cfd: "$formation.cfd",
+            rncp: "$formation.rncp",
+            id_erp_apprenant: "$id_erp_apprenant",
+            organisme_id: "$organisme_id",
+          },
+          count: {
+            $sum: 1,
+          },
+          docs: {
+            $addToSet: {
+              id: "$_id",
+              updated_at: "$updated_at",
+            },
+          },
+        },
+      },
+      {
+        $match: { count: { $gt: 1 } },
+      },
+    ])
+    .toArray();
+
+  for (const effectifDuplicat of effectifDuplicats) {
+    const { docs } = effectifDuplicat;
+    const lastUpdatedDoc = docs.reduce((acc, doc) => {
+      if (doc.updated_at.getTime() > acc.updated_at.getTime()) {
+        return doc;
+      }
+
+      return acc;
+    }, docs[0]);
+
+    await softDeleteEffectif(lastUpdatedDoc.id, null, {
+      motif: MOTIF_SUPPRESSION.Doublon,
+      description: "Suppression du doublon suite à la migration index unique sur effectifs",
+    });
+  }
+
+  const effectifDecaDuplicats = await effectifsDECADb()
+    .aggregate<{
+      _id: {
+        annee_scolaire: string;
+        cfd: string;
+        rncp: string;
+        id_erp_apprenant: string;
+        organisme_id: string;
+      };
+      count: number;
+      docs: {
+        id: ObjectId;
+        updated_at: Date;
+      }[];
+    }>([
+      {
+        $group: {
+          _id: {
+            annee_scolaire: "$annee_scolaire",
+            cfd: "$formation.cfd",
+            rncp: "$formation.rncp",
+            id_erp_apprenant: "$id_erp_apprenant",
+            organisme_id: "$organisme_id",
+          },
+          count: {
+            $sum: 1,
+          },
+          docs: {
+            $addToSet: {
+              id: "$_id",
+              updated_at: "$updated_at",
+            },
+          },
+        },
+      },
+      {
+        $match: { count: { $gt: 1 } },
+      },
+    ])
+    .toArray();
+
+  for (const effectifDuplicat of effectifDecaDuplicats) {
+    const { docs } = effectifDuplicat;
+    const lastUpdatedDoc = docs.reduce((acc, doc) => {
+      if (doc.updated_at.getTime() > acc.updated_at.getTime()) {
+        return doc;
+      }
+
+      return acc;
+    }, docs[0]);
+
+    await effectifsDECADb().deleteMany({ _id: { $ne: lastUpdatedDoc.id } });
+  }
+
+  // DROP unique index
+  await effectifsDb().dropIndex(
+    "organisme_id_1_annee_scolaire_1_id_erp_apprenant_1_apprenant.nom_1_apprenant.prenom_1_formation.cfd_1_formation.annee_1"
+  );
+};
diff --git a/server/src/jobs/hydrate/deca/hydrate-deca-raw.ts b/server/src/jobs/hydrate/deca/hydrate-deca-raw.ts
@@ -1,7 +1,7 @@
 import { normalize } from "path";
 
 import { captureException } from "@sentry/node";
-import { MongoClient, ObjectId, WithoutId } from "mongodb";
+import { MongoClient, MongoError, ObjectId, WithoutId } from "mongodb";
 import { SOURCE_APPRENANT } from "shared/constants";
 import { IOrganisme } from "shared/models";
 import { IRawBalDeca } from "shared/models/data/airbyteRawBalDeca.model";
@@ -96,20 +96,35 @@ export async function hydrateDecaRaw() {
   }
 }
 
-async function updateEffectifDeca(document: IRawBalDeca, count: { created: number; updated: number }) {
-  const newDocuments: WithoutId<IEffectifDECA>[] = await transformDocument(document);
-  const organismesId = new Set<string>();
-  for (const newDocument of newDocuments) {
-    const effectifFound = await checkIfEffectifExists(newDocument, effectifsDECADb());
+async function upsertEffectifDeca(
+  effectif: WithoutId<IEffectifDECA>,
+  count: { created: number; updated: number },
+  retry: boolean = true
+) {
+  const effectifFound = await checkIfEffectifExists<IEffectifDECA>(effectif, effectifsDECADb());
 
-    if (!effectifFound) {
-      await effectifsDECADb().insertOne(newDocument as IEffectifDECA);
+  if (!effectifFound) {
+    try {
+      await effectifsDECADb().insertOne({ ...effectif, _id: new ObjectId() });
       count.created++;
-    } else {
-      await effectifsDECADb().updateOne({ _id: effectifFound._id }, { $set: newDocument });
-      count.updated++;
+    } catch (err) {
+      // Le code d'erreur 11000 correspond à une duplication d'index unique
+      // Ce cas arrive lors du traitement concurrentiel du meme effectif dans la queue
+      if (retry && err instanceof MongoError && err.code === 11000) {
+        return upsertEffectifDeca(effectif, count, false);
+      }
     }
+  } else {
+    await effectifsDECADb().updateOne({ _id: effectifFound._id }, { $set: effectif });
+    count.updated++;
+  }
+}
 
+async function updateEffectifDeca(document: IRawBalDeca, count: { created: number; updated: number }) {
+  const newDocuments: WithoutId<IEffectifDECA>[] = await transformDocument(document);
+  const organismesId = new Set<string>();
+  for (const newDocument of newDocuments) {
+    await upsertEffectifDeca(newDocument, count);
     organismesId.add(newDocument.organisme_id.toString());
   }
 

diff --git a/server/src/jobs/ingestion/process-ingestion.ts b/server/src/jobs/ingestion/process-ingestion.ts
@@ -490,7 +490,7 @@ const createOrUpdateEffectif = async (
     },
   };
   const itemProcessingInfos: ItemProcessingInfos = {};
-  let effectifDb = await checkIfEffectifExists(effectifWithComputedFields, effectifsDb());
+  let effectifDb = await checkIfEffectifExists<IEffectif>(effectifWithComputedFields, effectifsDb());
   itemProcessingInfos.effectif_new = !effectifDb;
 
   try {

diff --git a/shared/models/data/effectifs.model.ts b/shared/models/data/effectifs.model.ts
@@ -26,10 +26,8 @@ const indexes: [IndexSpecification, CreateIndexesOptions][] = [
       organisme_id: 1,
       annee_scolaire: 1,
       id_erp_apprenant: 1,
-      "apprenant.nom": 1,
-      "apprenant.prenom": 1,
       "formation.cfd": 1,
-      "formation.annee": 1,
+      "formation.rncp": 1,
     },
     { unique: true },
   ],

diff --git a/shared/models/data/effectifsDECA.model.ts b/shared/models/data/effectifsDECA.model.ts
@@ -22,12 +22,15 @@ const collectionName = "effectifsDECA";
 const indexes: [IndexSpecification, CreateIndexesOptions][] = [
   [
     {
+      organisme_id: 1,
       annee_scolaire: 1,
-      "formation.cfd": 1,
       id_erp_apprenant: 1,
-      organisme_id: 1,
+      "formation.cfd": 1,
+      "formation.rncp": 1,
+    },
+    {
+      unique: true,
     },
-    {},
   ],
   [
     {