feat(import): add async CSV import queue with job status endpoint

.eslintrc.js

@@ -92,6 +92,9 @@ module.exports = {
     TTimeSeries: 'readonly',
     TTimeSeriesQualityLog: 'readonly',
     TUnit: 'readonly',
+    TJobBatch: 'readonly',
+    FileService: 'readonly',
+    CSVImportQueueService: 'readonly',
     VDataQualityComputeEntrance: 'readonly',
     sails: 'readonly',
   },

.gitignore

@@ -132,6 +132,9 @@ build-info.json
 # Temporary test output
 *.tmp
 
+# Local file uploads (dev environment, replaces Azure Blob Storage)
+.local-uploads/
+
 # Test database snapshot stamp
 test/.snapshot-stamp
 

api/controllers/v1/entrance/import-rows.js

@@ -1,20 +1,9 @@
-const CaveService = require('../../../services/CaveService');
-const CoordinatesSnapshotService = require('../../../services/CoordinatesSnapshotService');
-const EntranceService = require('../../../services/EntranceService');
-const EntranceCSVImportService = require('../../../services/EntranceCSVImportService');
 const RightService = require('../../../services/RightService');
 const {
   checkColumns,
-  valIfTruthyOrNull,
-  getOrCreateAuthor,
+  ENTRANCE_MANDATORY_COLUMNS,
 } = require('../../../utils/csvHelper');
 
-const ENTRANCE_MANDATORY_COLUMNS = [
-  'w3geo:latitude',
-  'w3geo:longitude',
-  'rdfs:label/dc:language',
-];
-
 module.exports = async (req, res) => {
   const hasRight = RightService.hasGroup(
     req.token.groups,
@@ -24,137 +13,43 @@ module.exports = async (req, res) => {
     return res.forbidden('You are not authorized to import entrances via CSV.');
   }
 
-  const requestResponse = {
-    type: 'entrance',
-    total: {
-      success: 0,
-      failure: 0,
-    },
-    successfulImport: [],
-    successfulImportAsDuplicates: [],
-    failureImport: [],
-  };
-
-  /* eslint-disable no-await-in-loop */
-  for (const [index, data] of req.body.data.entries()) {
-    const missingColumns = await checkColumns(data, ENTRANCE_MANDATORY_COLUMNS);
-
-    // Stop if missing columnes
-    if (missingColumns.length > 0) {
-      requestResponse.failureImport.push({
-        line: index + 2,
-        message: `Columns missing : ${missingColumns.toString()}`,
-      });
-      continue; // eslint-disable-line no-continue
-    }
-
-    // Check for duplicates
-    const idDb = valIfTruthyOrNull(data.id);
-    const nameDb = valIfTruthyOrNull(data['dct:rights/cc:attributionName']);
-
-    try {
-      // Get data
-      // Author retrieval: create one if not present in db
-      const authorId = await getOrCreateAuthor(data);
-      const dataNameDescLoc =
-        await EntranceCSVImportService.getConvertedNameDescLocEntranceFromCsv(
-          data,
-          authorId
-        );
-
-      const result = await TEntrance.findOne({
-        idDbImport: idDb,
-        nameDbImport: nameDb,
-      });
-      if (result) {
-        // Create a duplicate in DB
-        const cave = await TCave.findOne(result.cave);
-        const entrance = EntranceCSVImportService.getConvertedEntranceFromCsv(
-          data,
-          authorId,
-          cave
-        );
-
-        await TEntranceDuplicate.create({
-          author: req.token.id,
-          content: {
-            entrance,
-            nameDescLoc: dataNameDescLoc,
-          },
-          dateInscription: new Date(),
-          entrance: result.id,
-        });
-
-        requestResponse.successfulImportAsDuplicates.push({
-          line: index + 2,
-          message: `Entrance with id ${idDb} has been created as an entrance duplicate.`,
-        });
-        continue; // eslint-disable-line no-continue
-      }
-
-      // Cave creation
-      const dataCave = EntranceCSVImportService.getConvertedCaveFromCsv(
-        data,
-        authorId
-      );
-      const nameData =
-        EntranceCSVImportService.getConvertedNameAndDescCaveFromCsv(
-          data,
-          authorId
-        );
-      const createdCave = await CaveService.createCave(req, dataCave, nameData);
-
-      // Entrance creation
-      const dataEntrance = EntranceCSVImportService.getConvertedEntranceFromCsv(
-        data,
-        authorId,
-        createdCave
-      );
-      const { dateInscription } = dataEntrance;
-      const { dateReviewed } = dataEntrance;
+  const { data } = req.body || {};
+  if (!Array.isArray(data) || data.length === 0) {
+    return res.badRequest(
+      'Request body must contain a non-empty "data" array.'
+    );
+  }
 
-      const createdEntrance = await EntranceService.createEntrance(
-        req,
-        dataEntrance,
-        dataNameDescLoc
-      );
-      if (valIfTruthyOrNull(data['gn:alternateName'])) {
-        await TName.create({
-          author: authorId,
-          entrance: createdEntrance.id,
-          dateInscription,
-          dateReviewed,
-          isMain: false,
-          language: dataNameDescLoc.name.language,
-          name: data['gn:alternateName'].name,
-        });
-      }
+  // Fast-fail: validate mandatory columns on first row
+  const missingColumns = await checkColumns(
+    data[0],
+    ENTRANCE_MANDATORY_COLUMNS
+  );
+  if (missingColumns.length > 0) {
+    return res.badRequest(`Columns missing: ${missingColumns.toString()}`);
+  }
 
-      requestResponse.successfulImport.push({
-        caveId: createdCave.id,
-        entranceId: createdEntrance.id,
-        latitude: createdEntrance.latitude,
-        longitude: createdEntrance.longitude,
-      });
-    } catch (err) {
-      sails.log.error(err);
-      requestResponse.failureImport.push({
-        line: index + 2,
-        message: err.toString(),
-      });
-    }
+  if (!sails.enrichmentBoss) {
+    return res.serverError(
+      'Job queue is not available. Please try again later.'
+    );
   }
-  /* eslint-enable no-await-in-loop */
 
-  requestResponse.total.success = requestResponse.successfulImport.length;
-  requestResponse.total.successfulImportAsDuplicates =
-    requestResponse.successfulImportAsDuplicates.length;
-  requestResponse.total.failure = requestResponse.failureImport.length;
+  try {
+    const { batchId, totalRows, totalChunks } =
+      await CSVImportQueueService.createBatch(data, {
+        id: req.token.id,
+        groups: req.token.groups,
+      });
 
-  // Invalidate coordinates snapshot so newly imported entrances appear on the map
-  if (requestResponse.successfulImport.length > 0) {
-    CoordinatesSnapshotService.invalidate();
+    return res.status(202).json({
+      batchId,
+      totalRows,
+      totalChunks,
+      statusUrl: `/api/v1/jobs/${batchId}`,
+    });
+  } catch (err) {
+    sails.log.error('import-rows: failed to create batch:', err);
+    return res.serverError('Failed to enqueue import batch.');
   }
-
-  return res.ok(requestResponse);
 };

api/controllers/v1/job/find.js

@@ -0,0 +1,35 @@
+const RightService = require('../../../services/RightService');
+
+module.exports = async (req, res) => {
+  const { batchId } = req.params;
+
+  const batch = await TJobBatch.findOne({ id: batchId });
+  if (!batch) {
+    return res.notFound('Job batch not found.');
+  }
+
+  // Access control: users see only their own jobs unless moderator/admin
+  const isModerator = RightService.hasGroup(
+    req.token.groups,
+    RightService.G.MODERATOR
+  );
+  const isAdmin = RightService.hasGroup(
+    req.token.groups,
+    RightService.G.ADMINISTRATOR
+  );
+  if (batch.initiator !== req.token.id && !isModerator && !isAdmin) {
+    return res.forbidden('You are not authorized to view this job.');
+  }
+
+  const progress = await CSVImportQueueService.getBatchProgress(batchId);
+
+  return res.ok({
+    batchId: batch.id,
+    type: batch.type,
+    status: batch.status,
+    createdAt: batch.createdAt,
+    completedAt: batch.completedAt,
+    progress,
+    result: batch.result || null,
+  });
+};

api/models/TJobBatch.js

@@ -0,0 +1,77 @@
+/**
+ * TJobBatch.js
+ *
+ * @description :: Stores metadata for async job batches (e.g., CSV import).
+ */
+
+module.exports = {
+  tableName: 't_job_batch',
+
+  primaryKey: 'id',
+
+  attributes: {
+    id: {
+      type: 'string',
+      required: true,
+      columnName: 'id',
+    },
+
+    type: {
+      type: 'string',
+      required: true,
+      columnName: 'type',
+      maxLength: 50,
+    },
+
+    status: {
+      type: 'string',
+      required: true,
+      columnName: 'status',
+      isIn: ['pending', 'active', 'aggregating', 'completed', 'failed'],
+      maxLength: 20,
+    },
+
+    initiator: {
+      columnName: 'id_initiator',
+      model: 'TCaver',
+      required: true,
+    },
+
+    createdAt: {
+      type: 'ref',
+      columnType: 'timestamp',
+      columnName: 'created_at',
+      autoCreatedAt: true,
+    },
+
+    completedAt: {
+      type: 'ref',
+      columnType: 'timestamp',
+      columnName: 'completed_at',
+    },
+
+    totalRows: {
+      type: 'number',
+      required: true,
+      columnName: 'total_rows',
+    },
+
+    chunkSize: {
+      type: 'number',
+      required: true,
+      columnName: 'chunk_size',
+    },
+
+    totalChunks: {
+      type: 'number',
+      required: true,
+      columnName: 'total_chunks',
+    },
+
+    result: {
+      type: 'json',
+      columnName: 'result',
+      columnType: 'jsonb',
+    },
+  },
+};

api/models/TNotification.js

@@ -98,5 +98,10 @@ module.exports = {
       columnName: 'id_rigging',
       model: 'TRigging',
     },
+
+    jobBatch: {
+      columnName: 'id_job_batch',
+      model: 'TJobBatch',
+    },
   },
 };