[TECH] Enregistrer l'historique des réplications des données froides (PIX-16710)(PIX-16712).

api/.ls-lint.yml

@@ -29,6 +29,7 @@ ls:
 ignore:
   - .git
   - node_modules
+  - datamart/migrations
   - db/migrations
   - db/seeds
   - scripts

api/Procfile-maddo

@@ -4,4 +4,5 @@ postdeploy: npm run postdeploy:maddo
 # see https://github.com/1024pix/pix/pull/796
 # and https://github.com/npm/npm/issues/4603
 # for more information
-web: exec node maddo.js
+web: exec node index.maddo.js
+worker: exec node worker.js maddo

api/datamart/migrations/20250227142721_create-table-sco_certifications_results.js

@@ -0,0 +1,48 @@
+const TABLE_NAME = 'sco_certification_results';
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+const up = async function (knex) {
+  await knex.schema.createTable(TABLE_NAME, function (table) {
+    table.string('national_student_id');
+    table.index('national_student_id');
+    table.string('organization_uai');
+    table.string('last_name');
+    table.string('first_name');
+    table.date('birthdate');
+    table.text('status');
+    table.integer('pix_score');
+    table.timestamp('certification_date');
+    table.integer('competence_level');
+    table.index('organization_uai');
+    table.string('competence_name');
+    table.string('competence_code');
+    table.string('area_name');
+    table.string('certification_courses_id');
+  });
+  await knex.schema.raw('drop collation if exists "sco_certification_results_case_accent_punctuation_insensitive"');
+  await knex.schema.raw(
+    'create collation "sco_certification_results_case_accent_punctuation_insensitive" (provider = icu, locale = "und-u-ka-shifted-ks-level1-kv-punct", deterministic = false);',
+  );
+
+  await knex.schema.raw(
+    'ALTER TABLE :tableName: alter column first_name type varchar(255) COLLATE sco_certification_results_case_accent_punctuation_insensitive;',
+    { tableName: TABLE_NAME },
+  );
+  await knex.schema.raw(
+    'ALTER TABLE :tableName: alter column last_name type varchar(255) COLLATE sco_certification_results_case_accent_punctuation_insensitive;',
+    { tableName: TABLE_NAME },
+  );
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+const down = async function (knex) {
+  await knex.schema.dropTable(TABLE_NAME);
+};
+
+export { down, up };

api/datamart/migrations/20250227142903_create-table-certification_results.js

@@ -0,0 +1,35 @@
+const TABLE_NAME = 'certification_results';
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+const up = async function (knex) {
+  await knex.schema.createTable(TABLE_NAME, function (table) {
+    table.string('certification_code_verification');
+    table.string('last_name');
+    table.string('first_name');
+    table.date('birthdate');
+    table.text('status');
+    table.integer('pix_score');
+    table.timestamp('certification_date');
+    table.integer('competence_level');
+    table.index('certification_code_verification');
+    table.string('competence_name');
+    table.string('competence_code');
+    table.string('area_name');
+    table.string('certification_courses_id');
+    table.string('national_student_id');
+    table.string('organization_uai');
+  });
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+const down = async function (knex) {
+  await knex.schema.dropTable(TABLE_NAME);
+};
+
+export { down, up };

api/db/seeds/data/common/common-builder.js

@@ -85,7 +85,7 @@ function createClientApplications(databaseBuilder) {
     name: 'pixData',
     clientId: 'pixData',
     clientSecret: 'pixdatasecret',
-    scopes: ['statistics'],
+    scopes: ['statistics', 'replication'],
   });
   databaseBuilder.factory.buildClientApplication({
     name: 'parcoursup',

api/index.js

@@ -3,12 +3,14 @@ import 'dotenv/config';
 import { databaseConnections } from './db/database-connections.js';
 import { databaseConnection as liveDatabaseConnection } from './db/knex-database-connection.js';
 import { createServer } from './server.js';
+import { JobGroup } from './src/shared/application/jobs/job-controller.js';
 import { config, schema as configSchema } from './src/shared/config.js';
 import { learningContentCache } from './src/shared/infrastructure/caches/learning-content-cache.js';
 import { quitAllStorages } from './src/shared/infrastructure/key-value-storages/index.js';
 import { logger } from './src/shared/infrastructure/utils/logger.js';
 import { redisMonitor } from './src/shared/infrastructure/utils/redis-monitor.js';
 import { validateEnvironmentVariables } from './src/shared/infrastructure/validate-environment-variables.js';
+import { registerJobs } from './worker.js';
 
 validateEnvironmentVariables(configSchema);
 
@@ -62,7 +64,7 @@ process.on('SIGINT', () => {
   try {
     await start();
     if (config.infra.startJobInWebProcess) {
-      import('./worker.js');
+      registerJobs({ jobGroups: [JobGroup.DEFAULT, JobGroup.FAST] });
     }
   } catch (error) {
     logger.error(error);

api/maddo.js → api/index.maddo.js

@@ -2,11 +2,13 @@ import 'dotenv/config';
 
 import { databaseConnections } from './db/database-connections.js';
 import { createMaddoServer } from './server.maddo.js';
-import { schema as configSchema } from './src/shared/config.js';
+import { JobGroup } from './src/shared/application/jobs/job-controller.js';
+import { config, schema as configSchema } from './src/shared/config.js';
 import { quitAllStorages } from './src/shared/infrastructure/key-value-storages/index.js';
 import { logger } from './src/shared/infrastructure/utils/logger.js';
 import { redisMonitor } from './src/shared/infrastructure/utils/redis-monitor.js';
 import { validateEnvironmentVariables } from './src/shared/infrastructure/validate-environment-variables.js';
+import { registerJobs } from './worker.js';
 
 validateEnvironmentVariables(configSchema);
 
@@ -45,6 +47,9 @@ process.on('SIGINT', () => {
 (async () => {
   try {
     await start();
+    if (config.infra.startJobInWebProcess) {
+      registerJobs({ jobGroups: [JobGroup.MADDO] });
+    }
   } catch (error) {
     logger.error(error);
     throw error;

api/package.json

@@ -172,7 +172,7 @@
     "scalingo-postbuild": "node scripts/generate-cron > cron.json && node scripts/generate-procfile",
     "dev": "nodemon index.js",
     "start": "node index.js",
-    "start:maddo": "MADDO=true node maddo.js",
+    "start:maddo": "MADDO=true node index.maddo.js",
     "start:watch": "npm run dev",
     "start:job": "node worker.js",
     "start:job:fast": "node worker.js fast",

api/server.maddo.js

@@ -4,6 +4,7 @@ import { parse } from 'neoqs';
 
 import { setupErrorHandling } from './config/server-setup-error-handling.js';
 import { knex } from './db/knex-database-connection.js';
+import * as authenticationRoutes from './lib/application/authentication/index.js';
 import { authentication } from './lib/infrastructure/authentication.js';
 import * as replicationRoutes from './src/maddo/application/replications-routes.js';
 import { Metrics } from './src/monitoring/infrastructure/metrics.js';
@@ -178,7 +179,7 @@ const setupAuthentication = function (server) {
 };
 
 const setupRoutesAndPlugins = async function (server) {
-  await server.register([...plugins, healthcheckRoutes, replicationRoutes]);
+  await server.register([...plugins, healthcheckRoutes, authenticationRoutes, replicationRoutes]);
 };
 
 const setupOpenApiSpecification = async function (server) {

api/src/maddo/application/jobs/replication-job-controller.js

@@ -0,0 +1,25 @@
+import { knex as datamartKnex } from '../../../../datamart/knex-database-connection.js';
+import { knex as datawarehouseKnex } from '../../../../datawarehouse/knex-database-connection.js';
+import { JobController, JobGroup } from '../../../shared/application/jobs/job-controller.js';
+import { ReplicationJob } from '../../domain/models/ReplicationJob.js';
+import { extractTransformAndLoadData } from '../../domain/usecases/extract-transform-and-load-data.js';
+import * as replicationRepository from '../../infrastructure/repositories/replication-repository.js';
+
+export class ReplicationJobController extends JobController {
+  constructor() {
+    super(ReplicationJob.name, { jobGroup: JobGroup.MADDO });
+  }
+
+  async handle({
+    data: { replicationName },
+    dependencies = { extractTransformAndLoadData, replicationRepository, datamartKnex, datawarehouseKnex },
+  }) {
+    const { extractTransformAndLoadData, replicationRepository, datamartKnex, datawarehouseKnex } = dependencies;
+    return extractTransformAndLoadData({
+      replicationName,
+      replicationRepository,
+      datamartKnex,
+      datawarehouseKnex,
+    });
+  }
+}

api/src/maddo/application/replications-controller.js

@@ -1,47 +1,25 @@
-import { knex as datamartKnex } from '../../../datamart/knex-database-connection.js';
-import { knex as datawarehouseKnex } from '../../../datawarehouse/knex-database-connection.js';
-import { logger } from '../../shared/infrastructure/utils/logger.js';
-import { extractTransformAndLoadData } from '../domain/usecases/extract-transform-and-load-data.js';
+import { ReplicationJob } from '../domain/models/ReplicationJob.js';
+import { replicationJobRepository } from '../infrastructure/repositories/jobs/replication-job-repository.js';
 import * as replicationRepository from '../infrastructure/repositories/replication-repository.js';
 
 export async function replicate(
   request,
   h,
   dependencies = {
-    extractTransformAndLoadData,
     replicationRepository,
-    datamartKnex,
-    datawarehouseKnex,
-    logger,
+    replicationJobRepository,
   },
 ) {
+  const { replicationRepository, replicationJobRepository } = dependencies;
   const { replicationName } = request.params;
 
-  const replication = dependencies.replicationRepository.getByName(replicationName);
+  const replication = replicationRepository.getByName(replicationName);
 
   if (!replication) {
     return h.response().code(404);
   }
 
-  const promise = dependencies
-    .extractTransformAndLoadData({
-      replication,
-      datamartKnex: dependencies.datamartKnex,
-      datawarehouseKnex: dependencies.datawarehouseKnex,
-    })
-    .catch((err) =>
-      dependencies.logger.error(
-        {
-          event: 'replication',
-          err,
-        },
-        'Error during replication',
-      ),
-    );
-
-  if (!request.query.async) {
-    await promise;
-  }
+  await replicationJobRepository.performAsync(new ReplicationJob({ replicationName }));
 
   return h.response().code(204);
 }

api/src/maddo/domain/models/ReplicationJob.js

@@ -0,0 +1,5 @@
+export class ReplicationJob {
+  constructor({ replicationName }) {
+    this.replicationName = replicationName;
+  }
+}

api/src/maddo/domain/usecases/extract-transform-and-load-data.js

@@ -1,6 +1,13 @@
 const DEFAULT_CHUNK_SIZE = 1_000;
 
-export async function extractTransformAndLoadData({ replication, datawarehouseKnex, datamartKnex }) {
+export async function extractTransformAndLoadData({
+  replicationName,
+  replicationRepository,
+  datawarehouseKnex,
+  datamartKnex,
+}) {
+  const replication = replicationRepository.getByName(replicationName);
+
   let context = { datawarehouseKnex, datamartKnex };
 
   const additionnalContext = await replication.before?.(context);
@@ -9,11 +16,13 @@ export async function extractTransformAndLoadData({ replication, datawarehouseKn
 
   const queryBuilder = replication.from(context);
   let chunk = [];
+  let count = 0;
   const chunkSize = replication.chunkSize ?? DEFAULT_CHUNK_SIZE;
   const connection = await datamartKnex.context.client.acquireConnection();
   try {
     for await (const data of queryBuilder.stream()) {
       chunk.push(replication.transform?.(data) ?? data);
+      count += 1;
       if (chunk.length === chunkSize) {
         await replication.to(context, chunk).connection(connection);
         chunk = [];
@@ -23,6 +32,7 @@ export async function extractTransformAndLoadData({ replication, datawarehouseKn
     if (chunk.length) {
       await replication.to(context, chunk).connection(connection);
     }
+    return { count };
   } finally {
     await datamartKnex.context.client.releaseConnection(connection);
   }

api/src/maddo/infrastructure/repositories/jobs/replication-job-repository.js

@@ -0,0 +1,4 @@
+import { JobRepository, JobRetry } from '../../../../shared/infrastructure/repositories/jobs/job-repository.js';
+import { ReplicationJob } from '../../../domain/models/ReplicationJob.js';
+
+export const replicationJobRepository = new JobRepository({ name: ReplicationJob.name, retry: JobRetry.FEW_RETRY });

api/src/maddo/infrastructure/repositories/replication-repository.js

@@ -1,4 +1,53 @@
-export const replications = [];
+export const replications = [
+  {
+    name: 'sco_certification_results',
+    before: async ({ datamartKnex }) => {
+      await datamartKnex('sco_certification_results').delete();
+    },
+    from: ({ datawarehouseKnex }) => {
+      return datawarehouseKnex('data_export_parcoursup_certif_result').select('*');
+    },
+    to: ({ datamartKnex }, chunk) => {
+      return datamartKnex('sco_certification_results').insert(chunk);
+    },
+  },
+  {
+    name: 'data_export_parcoursup_certif_result',
+    before: async ({ datamartKnex }) => {
+      await datamartKnex('data_export_parcoursup_certif_result').delete();
+    },
+    from: ({ datawarehouseKnex }) => {
+      return datawarehouseKnex('data_export_parcoursup_certif_result').select('*');
+    },
+    to: ({ datamartKnex }, chunk) => {
+      return datamartKnex('data_export_parcoursup_certif_result').insert(chunk);
+    },
+  },
+  {
+    name: 'certification_results',
+    before: async ({ datamartKnex }) => {
+      await datamartKnex('certification_results').delete();
+    },
+    from: ({ datawarehouseKnex }) => {
+      return datawarehouseKnex('data_export_parcoursup_certif_result_code_validation').select('*');
+    },
+    to: ({ datamartKnex }, chunk) => {
+      return datamartKnex('certification_results').insert(chunk);
+    },
+  },
+  {
+    name: 'data_export_parcoursup_certif_result_code_validation',
+    before: async ({ datamartKnex }) => {
+      await datamartKnex('data_export_parcoursup_certif_result_code_validation').delete();
+    },
+    from: ({ datawarehouseKnex }) => {
+      return datawarehouseKnex('data_export_parcoursup_certif_result_code_validation').select('*');
+    },
+    to: ({ datamartKnex }, chunk) => {
+      return datamartKnex('data_export_parcoursup_certif_result_code_validation').insert(chunk);
+    },
+  },
+];
 
 export function getByName(name, dependencies = { replications }) {
   return dependencies.replications.find((replication) => replication.name === name);

api/src/monitoring/infrastructure/metrics.js

@@ -1,6 +1,8 @@
 import metrics from 'datadog-metrics';
 
-import { logger } from '../../shared/infrastructure/utils/logger.js';
+import { child } from '../../shared/infrastructure/utils/logger.js';
+
+const logger = child('metrics', { event: 'metrics' });
 
 export class Metrics {
   static intervals = [];

api/src/shared/application/jobs/job-controller.js

@@ -7,6 +7,7 @@ import { JobExpireIn } from '../../infrastructure/repositories/jobs/job-reposito
 export const JobGroup = {
   DEFAULT: 'default',
   FAST: 'fast',
+  MADDO: 'maddo',
 };
 
 export class JobController {