fixed data import validation job (error message) and job progress not…

…ification

core/functionsDefer.js

@@ -3,10 +3,10 @@ const throttleTimeouts = {}
 const throttleLastRan = {}
 
 export const debounce = (func, id, delay = 500, immediate = false) => {
-  return function(...args) {
+  return function (...args) {
     const context = this
 
-    const later = function() {
+    const later = function () {
       delete debounceTimeouts[id]
       if (!immediate) {
         func.apply(context, args)
@@ -23,7 +23,7 @@ export const debounce = (func, id, delay = 500, immediate = false) => {
 }
 
 export const throttle = (func, id, limit = 500) => {
-  return function(...args) {
+  return function (...args) {
     const context = this
 
     const runFunction = () => {
@@ -37,18 +37,25 @@ export const throttle = (func, id, limit = 500) => {
     if (lastRun) {
       clearTimeout(throttleTimeouts[id])
 
-      throttleTimeouts[id] = setTimeout(() => {
-        if (Date.now() - lastRun >= limit) {
-          runFunction()
-        }
-      }, limit - (Date.now() - lastRun))
+      const timeSinceLastRun = Date.now() - lastRun
+      const nextRunTimeout = limit - timeSinceLastRun
+
+      if (nextRunTimeout > 0) {
+        throttleTimeouts[id] = setTimeout(() => {
+          if (Date.now() - lastRun >= limit) {
+            runFunction()
+          }
+        }, nextRunTimeout)
+      } else {
+        runFunction()
+      }
     } else {
       runFunction()
     }
   }
 }
 
-export const cancelThrottle = id => {
+export const cancelThrottle = (id) => {
   const timeout = throttleTimeouts[id]
   if (timeout) {
     clearTimeout(timeout)

core/i18n/resources/en.js

@@ -1464,7 +1464,7 @@ Levels will be renamed into level_1, level_2... level_N and an extra 'area' prop
       errorUpdatingValues: 'Error updating values',
       recordAlreadyExisting: 'Record with keys "{{keyValues}}" already existing',
       recordInAnalysisStepCannotBeUpdated: 'Record with keys "{{keyValues}}" is in Analysis step and cannot be updated',
-      recordKeysMissing: 'Missing record key value',
+      recordKeyMissing: 'Missing value for key attribute "{{keyName}}"',
       recordNotFound: 'Record with keys "{{keyValues}}" not found',
     },
 

core/validation/_validator/validatorErrorKeys.js

@@ -51,7 +51,7 @@ export const ValidatorErrorKeys = {
     emptyFile: 'validationErrors.dataImport.emptyFile',
     errorUpdatingValues: 'validationErrors.dataImport.errorUpdatingValues',
     recordAlreadyExisting: 'validationErrors.dataImport.recordAlreadyExisting',
-    recordKeysMissing: 'validationErrors.dataImport.recordKeysMissing',
+    recordKeyMissing: 'validationErrors.dataImport.recordKeyMissing',
     recordNotFound: 'validationErrors.dataImport.recordNotFound',
   },
 

server/job/job.js

@@ -213,7 +213,7 @@ export default class Job {
     this.processed += incrementBy
 
     throttle(
-      async () => await this._notifyEvent(this._createJobEvent(jobEvents.progress)),
+      async () => this._notifyEvent(this._createJobEvent(jobEvents.progress)),
       this._getProgressThrottleId(),
       1000
     )()

server/job/jobThread.js

@@ -19,7 +19,7 @@ class JobThread extends Thread {
         this.sendJobToParentThread()
         break
       case jobThreadMessageTypes.cancelJob:
-        this.job.cancel()
+        await this.job.cancel()
         break
       default:
         console.log(`Skipping unknown message type: ${msg.type}`)

server/modules/dataImport/service/DataImportJob/DataImportJob.js

@@ -13,6 +13,8 @@ import DataImportBaseJob from './DataImportBaseJob'
 export default class DataImportJob extends DataImportBaseJob {
   constructor(params, type = DataImportJob.type) {
     super(type, params)
+
+    this.csvReader = null
   }
 
   async execute() {
@@ -89,27 +91,37 @@ export default class DataImportJob extends DataImportBaseJob {
     const { cycle, entityDefUuid, filePath, survey } = this.context
 
     try {
-      const reader = await DataImportFileReader.createReader({
+      this.csvReader = await DataImportFileReader.createReader({
         filePath,
         survey,
         cycle,
         entityDefUuid,
-        onRowItem: (item) => this.onRowItem(item),
+        onRowItem: async (item) => this.onRowItem(item),
         onTotalChange: (total) => (this.total = total),
       })
 
-      await reader.start()
+      await this.csvReader.start()
     } catch (e) {
       const errorKey = e.key || e.toString()
       const errorParams = e.params
       this._addError(errorKey, errorParams)
     }
   }
 
+  async cancel() {
+    await super.cancel()
+
+    this.csvReader?.cancel()
+  }
+
   async onRowItem({ valuesByDefUuid, errors }) {
     const { context, tx } = this
     const { entityDefUuid, insertMissingNodes, survey } = context
 
+    if (this.isCanceled()) {
+      return
+    }
+
     this.incrementProcessedItems()
 
     errors.forEach((error) => {

server/modules/dataImport/service/DataImportJob/recordProvider.js

@@ -11,6 +11,14 @@ import * as Validation from '@core/validation/validation'
 
 import * as RecordManager from '@server/modules/record/manager/recordManager'
 
+const checkRootKeysSpecified = ({ rootKeyDefs, rootKeyValuesFormatted }) => {
+  const emptyRootKeyValueIndex = rootKeyValuesFormatted.findIndex(Objects.isEmpty)
+  if (emptyRootKeyValueIndex >= 0) {
+    const keyName = NodeDef.getName(rootKeyDefs[emptyRootKeyValueIndex])
+    throw new SystemError(Validation.messageKeys.dataImport.recordKeyMissing, { keyName })
+  }
+}
+
 const fetchOrCreateRecord = async ({ valuesByDefUuid, currentRecord, context, tx }) => {
   const { cycle, dryRun, insertNewRecords, recordsSummary, survey, surveyId, updateRecordsInAnalysis, user } = context
 
@@ -25,10 +33,7 @@ const fetchOrCreateRecord = async ({ valuesByDefUuid, currentRecord, context, tx
     })
   )
 
-  // check root keys are not empty
-  if (rootKeyValuesFormatted.some(Objects.isEmpty)) {
-    throw new SystemError(Validation.messageKeys.dataImport.recordKeysMissing)
-  }
+  checkRootKeysSpecified({ rootKeyDefs, rootKeyValuesFormatted })
 
   const recordSummary = recordsSummary.find((record) =>
     rootKeyDefs.every((rootKeyDef) => {