fix(task): merge map with 1 only

.gitignore

@@ -1,3 +1,4 @@
 .env
+.eslintcache
 node_modules
 yarn-error.log

k8s/kinto/job-init-kinto.yml

@@ -30,3 +30,18 @@ spec:
               cpu: ${INIT_KINTO_RESOURCE_CPU_LIMIT}
               memory: ${INIT_KINTO_RESOURCE_MEMORY_LIMIT}
       restartPolicy: Never
+      initContainers:
+        - name: wait-for-kinto
+          image: registry.gitlab.factory.social.gouv.fr/socialgouv/docker/curl:1.12.0
+          imagePullPolicy: Always
+          command:
+            - sh
+            - -c
+            - |
+              retry=300; # (60 x 1s) x 5min
+              while
+                ! curl -sSL "http://kinto:8888/__lbheartbeat__" &&
+                [[ $(( retry-- )) -gt 0 ]];
+              do echo "Waiting for ${API_URL} to be available ($(( retry )))" ; sleep 1s ; done ;
+              [[ $(( retry )) -lt 1 ]] && exit 128;
+              echo Ready;

packages/api/src/collector/service/dossier.service.ts

@@ -106,7 +106,7 @@ class DossierService {
           );
           return dossierRepository.update(loadedRecord.id || "0", loadedRecord);
         }
-      })
+      }, 1)
     );
   }
 }

packages/api/src/collector/service/task.service.ts

@@ -30,6 +30,9 @@ class TaskService {
   }
 
   public markAsCompleted(task: Task): Observable<Task> {
+    logger.info(
+      `[TaskService] task ${task.procedure_id}-${task.dossier_id} ${task.action} completed`
+    );
     task.task_completed_date = new Date().toISOString();
     task.task_state = "completed";
     return taskRepository.update(task);

packages/api/src/routes.ts

@@ -63,7 +63,7 @@ router.post(
       .pipe(
         map(() => dsConfigs),
         flatMap((x: ProcedureConfig[]) => x),
-        mergeMap((x: ProcedureConfig) => dsProcedureConfigRepository.add(x)),
+        mergeMap((x: ProcedureConfig) => dsProcedureConfigRepository.add(x), 1),
         tap((res: ProcedureConfig) =>
           logger.info(`ds config ${res.group.label} added`)
         )

packages/api/src/scheduler/dossier-synchro.service.ts

@@ -32,7 +32,7 @@ export const dossierSynchroService = {
           return dossierService.save(config.group, procId, dossier);
         }
         return dossierService.update(dossierRecord, dossier);
-      }),
+      }, 1),
       tap((dossier: DossierRecord | null) => {
         if (dossier) {
           logger.info(

packages/api/src/scheduler/dossier.scheduler.ts

@@ -1,5 +1,5 @@
-import { combineLatest, Observable } from "rxjs";
-import { concatMap, flatMap, map, mergeMap, tap } from "rxjs/operators";
+import { combineLatest, Observable, forkJoin } from "rxjs";
+import { concatMap, flatMap, map, mergeMap, tap, share } from "rxjs/operators";
 import {
   apiResultService,
   dsProcedureConfigService,
@@ -27,35 +27,44 @@ interface SyncContext {
 export const dossierScheduler = {
   start: () => {
     handleScheduler(configuration.schedulerCronDS, "dossier-synchro", () => {
-      return syncProcedures().pipe(
+      const apiResult$ = syncProcedures().pipe(
         mergeMap((x: ProcedureRecord) => buildSyncContext(x)),
         map((x: SyncContext) => {
           const actions = getSynchroActions(x.items, x.apiResult);
           x.apiResult.items = x.items;
           x.apiResult.actions = actions;
           return x.apiResult;
         }),
+        share()
+      );
+
+      const addAllTasks$ = apiResult$.pipe(
+        tap(apiResult =>
+          logger.info(
+            `[dossier.scheduler] procedure#${apiResult.procedure} - add ${apiResult.actions.length} actions`
+          )
+        ),
         flatMap((x: APIResult) =>
           x.actions.map((a: SynchroAction) => ({ action: a, apiResult: x }))
         ),
-        mergeMap(
-          (input: { action: SynchroAction; apiResult: APIResult }) => {
-            return taskService.addTask(
-              input.action.action,
-              input.action.procedure,
-              input.action.item.id,
-              input.action.item.state,
-              input.action.item.updated_at
-            );
-          },
-          (input: { action: SynchroAction; apiResult: APIResult }) =>
-            input.apiResult,
-          5
-        ),
+        mergeMap(({ action }) => {
+          return taskService.addTask(
+            action.action,
+            action.procedure,
+            action.item.id,
+            action.item.state,
+            action.item.updated_at
+          );
+        }, 1)
+      );
+
+      const updateApiResult$ = apiResult$.pipe(
         mergeMap((apiResult: APIResult) => {
           return apiResultService.update(apiResult);
-        })
+        }, 1)
       );
+
+      return forkJoin(addAllTasks$, updateApiResult$);
     });
   }
 };
@@ -111,6 +120,11 @@ function syncProcedures(): Observable<ProcedureRecord> {
 
 function allDemarcheSimlifieeProcedures(): Observable<DSProcedure> {
   return dsProcedureConfigService.all().pipe(
+    tap(procedureConfigs =>
+      logger.info(
+        `[SyncService.allDemarcheSimlifieeProcedures] process ${procedureConfigs.length} procedures`
+      )
+    ),
     flatMap((x: ProcedureConfig[]) => x),
     flatMap((x: ProcedureConfig) => x.procedures),
     concatMap(demarcheSimplifieeService.getDSProcedure),

packages/api/src/scheduler/scheduler.service.ts

@@ -34,7 +34,7 @@ export const handleScheduler = (
     runScheduler$
       .pipe(
         takeLast(1),
-        mergeMap(() => synchroHistoryService.update(scheduler, end))
+        mergeMap(() => synchroHistoryService.update(scheduler, end), 1)
       )
       .subscribe({
         complete: () => completeProcess(scheduler),

packages/api/src/scheduler/task.scheduler.ts

@@ -1,5 +1,13 @@
-import { combineLatest, Observable, of } from "rxjs";
-import { exhaustMap, flatMap, mergeMap, reduce } from "rxjs/operators";
+import { combineLatest, Observable, of, EMPTY } from "rxjs";
+import {
+  exhaustMap,
+  flatMap,
+  mergeMap,
+  reduce,
+  catchError,
+  tap,
+  take
+} from "rxjs/operators";
 import {
   dossierService,
   dsProcedureConfigService,
@@ -11,6 +19,7 @@ import { statisticService } from "../collector/service/statistic.service";
 import { configuration } from "../config";
 import { dossierSynchroService } from "./dossier-synchro.service";
 import { handleScheduler } from "./scheduler.service";
+import { logger } from "../util";
 
 export const taskScheduler = {
   start: () => {
@@ -19,10 +28,11 @@ export const taskScheduler = {
         allTasksToComplete(),
         dsProcedureConfigService.all()
       ).pipe(
+        take(50),
         mergeMap(
           ([task, procedures]) => processTask(task, procedures),
           undefined,
-          5
+          1
         ),
         reduce((acc: Task[], record: Task) => {
           acc.push(record);
@@ -61,12 +71,30 @@ function processTask(taskToTreat: Task, procedures: ProcedureConfig[]) {
           );
         }
       },
-      (task: Task) => task
+      (task: Task) => task,
+      1
     ),
-    mergeMap((task: Task) => taskService.markAsCompleted(task))
+    mergeMap(
+      (task: Task) =>
+        taskService.markAsCompleted(task).pipe(
+          catchError(err => {
+            logger.error(
+              `[task.scheduler] cannot update as completed task ${task.id}`,
+              err
+            );
+            return EMPTY;
+          })
+        ),
+      1
+    )
   );
 }
 
 function allTasksToComplete(): Observable<Task> {
-  return taskService.getTasksToComplete().pipe(flatMap((x: Task[]) => x));
+  return taskService.getTasksToComplete().pipe(
+    tap(tasks => {
+      logger.debug(`[task.scheduler] process ${tasks.length} to complete`);
+    }),
+    flatMap((x: Task[]) => x)
+  );
 }

packages/api/src/util/logger/logger.ts

@@ -59,6 +59,8 @@ const wLogger = createLogger({
 
 const logger = {
   debug: (message: string) => wLogger.debug(message),
+  profile: (id: string) => wLogger.profile(id),
+  startTimer: () => wLogger.startTimer(),
   error: (message: string, err: Error) => {
     wLogger.error(message, err);
     if (configuration.sentryEnabled) {

packages/kinto/src/index.js

@@ -42,28 +42,58 @@ const addDSConfig = async () => {
       },
       procedures: [6274, 6286]
     };
-    const res = await api.createRecord("ds_collector", "ds_configs", dsConfig);
-    console.log("[init kinto] add ds_configs record:  ", res);
+    {
+      const res = await api.createRecord(
+        "ds_collector",
+        "ds_configs",
+        dsConfig
+      );
+      console.log("[init kinto] add ds_configs record:  ", res);
+    }
+    {
+      const res = await api.createRecord("ds_collector", "synchro_histories", {
+        scheduler: "task",
+        last_synchro: 0
+      });
+      console.log("[init kinto] add synchro_histories record:  ", res);
+    }
   } else {
     console.log(
-      `[init kinto] ENVIRONMENT_TYPE=  "${
-        configs.environmentType
-      }, no 'ds_configs' record has been created.`
+      `[init kinto] ENVIRONMENT_TYPE=  "${configs.environmentType}, no 'ds_configs' record has been created.`
     );
   }
 };
-const init = async () => {
+const cleanIt = () => {
+  console.log("[haxxx] delete the admin account");
   await api
+    .deleteAdmin(configs.adminLogin, configs.adminPassword)
+    .catch(console.error);
+  const isNewUser = await api
     .createAdmin(configs.adminLogin, configs.adminPassword)
-    .then(async res => {
-      await addCollections();
-      if (res.data) {
-        console.log("[init kinto] admin created", res);
-        await addDSConfig();
-      } else {
-        console.log("[init kinto] kinto already initialised... ", res);
-      }
-    });
+    .then(
+      res => Boolean(res.data),
+      () => false
+    );
+
+  console.log("[haxxx] delete the ds_collector bucket");
+  try {
+    await api.deleteBucket("ds_collector");
+  } catch {}
+}
+
+const init = async () => {
+  if (process.env.CLEAN_DB) {
+    cleanIt();
+  }
+  console.log("[init kinto] addCollections");
+  await addCollections();
+
+  if (isNewUser) {
+    console.log("[init kinto] admin created");
+    await addDSConfig();
+  } else {
+    console.log("[init kinto] kinto already initialised... ");
+  }
 };
 
-init();
+init().catch(console.error);

packages/kinto/src/kinto-api.js

@@ -27,15 +27,27 @@ const header = auth => {
 
 const api = async (url, options) => {
   const response = await fetch(url, options);
+  if (!response.ok) throw response;
   return response.json();
 };
 
-module.exports.createAdmin = async function(login, password) {
+exports.createAdmin = async function(login, password) {
   const body = { data: { password: password } };
   return api(_account(login), _requestOptions("PUT", undefined, body));
 };
 
-module.exports.createUser = async function(login, password) {
+exports.deleteAdmin = async function(login, password) {
+  return api(
+    _account(login),
+    _requestOptions(
+      "DELETE",
+      `${configs.adminLogin}:${configs.adminPassword}`,
+      {}
+    )
+  );
+};
+
+exports.createUser = async function(login, password) {
   const body = { data: { password: password } };
   return api(
     _account(login),
@@ -47,7 +59,7 @@ module.exports.createUser = async function(login, password) {
   );
 };
 
-module.exports.createBucket = async function(name) {
+exports.createBucket = async function(name) {
   const body = { data: { id: name } };
   return api(
     _buckets(),
@@ -59,7 +71,18 @@ module.exports.createBucket = async function(name) {
   );
 };
 
-module.exports.createCollection = async function(bucket, collection) {
+exports.deleteBucket = async function(name) {
+  return api(
+    _bucket(name),
+    _requestOptions(
+      "DELETE",
+      `${configs.adminLogin}:${configs.adminPassword}`,
+      {}
+    )
+  );
+};
+
+exports.createCollection = async function(bucket, collection) {
   const body = { data: { id: collection } };
   return api(
     _collections(bucket),
@@ -71,7 +94,18 @@ module.exports.createCollection = async function(bucket, collection) {
   );
 };
 
-module.exports.createRecord = async function(bucket, collection, data) {
+exports.deleteCollection = async function(bucket, name) {
+  return api(
+    `${_collections(bucket)}/${name}`,
+    _requestOptions(
+      "DELETE",
+      `${configs.adminLogin}:${configs.adminPassword}`,
+      {}
+    )
+  );
+};
+
+exports.createRecord = async function(bucket, collection, data) {
   const body = { data };
   return api(
     _records(bucket, collection),