Source connector saves directly datapoints

Query-Engine/docker-compose.prod.yml

@@ -65,8 +65,6 @@ services:
       #- .:/workspace:cached
       command: sleep infinity
 
-
-
   libretranslate:
     image: libretranslate/libretranslate:latest
     container_name: libretranslate

Query-Engine/package.json

@@ -22,12 +22,13 @@
     "cors": "^2.8.5",
     "dotenv": "^16.4.5",
     "express": "^4.19.2",
-    "jsonwebtoken": "^9.0.2",
+    "ioredis": "^5.3.2",
+    "jsonwebtoken": "^9.0.3",
+    "jwa": "^2.0.1",
     "minio": "^8.0.0",
     "mongoose": "^8.3.4",
     "nodemon": "^3.1.0",
     "percocologger": "^1.0.7",
-    "ioredis": "^5.3.2",
     "pg": "^8.11.5"
   },
   "stashDependencies": {

Source-Connector/api/models/Datapoint.js

@@ -1,14 +1,94 @@
-const mongoose = require('mongoose');
-
-const datapointSchema = new mongoose.Schema({
-  source: String,
-  survey: String,
-  surveyName: String,
-  region: String,
-  fromUrl: String,
-  timestamp: String,
-  dimensions: Object,
-  value: Number
-}, {strict: false});
-
-module.exports = mongoose.model('Datapoint', datapointSchema);
+const mongoose = require("mongoose");
+const crypto = require("crypto");
+
+// Funzione di utilità per pulire il nome della survey
+const cleanSurveyName = (val) => {
+  if (!val || typeof val !== "string") return val;
+  return val.toUpperCase().replace(/\./g, ""); // Toglie i punti e mette in MAIUSCOLO
+};
+
+const datapointSchema = new mongoose.Schema(
+  {
+    source: { type: String, index: true, uppercase: true }, // uppercase: true lo fa in automatico
+    survey: { type: String, index: true, uppercase: true },
+    surveyName: String,
+    region: { type: String, index: true, uppercase: true },
+    fromUrl: String,
+    timestamp: { type: String, index: true },
+    dimensions: { type: Object },
+    value: Number,
+    dupl_hash: { type: String, unique: true }, // Identificatore unico per evitare duplicati
+  },
+  {
+    strict: false,
+    timestamps: true,
+  }
+);
+
+// Indici per le performance
+datapointSchema.index({ "dimensions.$**": 1 });
+datapointSchema.index({ region: 1, timestamp: -1 });
+
+// Middleware: Prima di ogni salvataggio, pulisce la survey
+datapointSchema.pre("save", function (next) {
+  if (this.survey) this.survey = cleanSurveyName(this.survey);
+  next();
+});
+
+// Funzione per generare l'hash unico (usata nell'inserimento massivo)
+const generateHash = (doc) => {
+  // Gestiamo sia se arriva come documento Mongoose sia come oggetto puro
+  const target = doc._doc || doc;
+
+  const dims = target.dimensions || {};
+  const sortedKeys = Object.keys(dims)
+    .sort()
+    .reduce((acc, key) => {
+      acc[key] = dims[key];
+      return acc;
+    }, {});
+
+  const s = cleanSurveyName(target.survey);
+
+  // COSTRUZIONE STRINGA DI HASH
+  const stringToHash =
+    (target.fromUrl || "") +
+    "|" + 
+    (target.timestamp || "") +
+    "|" +
+    (target.region || "") +
+    "|" + 
+    s +
+    "|" +
+    (target.value !== undefined ? target.value : Date.now()) +
+    "|" + 
+    JSON.stringify(sortedKeys);
+
+  return crypto.createHash("md5").update(stringToHash).digest("hex");
+};
+
+// Metodo per inserire milioni di record senza duplicati
+datapointSchema.statics.upsertMany = async function (datapoints) {
+  if (!datapoints || datapoints.length === 0) return;
+
+  const operations = datapoints.map((doc) => {
+    const hash = generateHash(doc);
+    // Puliamo la survey anche qui per l'operazione di bulk
+    const cleanedDoc = {
+      ...doc,
+      survey: cleanSurveyName(doc.survey),
+      dupl_hash: hash,
+    };
+    return {
+      updateOne: {
+        filter: { dupl_hash: hash },
+        update: { $set: cleanedDoc },
+        upsert: true,
+      },
+    };
+  });
+
+  return this.bulkWrite(operations, { ordered: false });
+};
+
+module.exports = mongoose.model("Datapoint", datapointSchema);