feat: WIP implementing new model of raw responses

Author: loichenninger

src/main/java/fr/insee/genesis/controller/dto/ScheduleDto.java

@@ -8,6 +8,7 @@
 
 @Builder
 public record ScheduleDto (String surveyName,
+						   String collectionInstrumentId,
 						   LocalDateTime lastExecution,
 						   List<KraftwerkExecutionSchedule> kraftwerkExecutionScheduleList
 ){}

src/main/java/fr/insee/genesis/controller/rest/responses/RawResponseController.java

@@ -13,6 +13,7 @@
 import fr.insee.genesis.domain.model.surveyunit.rawdata.DataProcessResult;
 import fr.insee.genesis.domain.model.surveyunit.rawdata.LunaticJsonRawDataModel;
 import fr.insee.genesis.domain.ports.api.LunaticJsonRawDataApiPort;
+import fr.insee.genesis.domain.ports.api.RawResponseApiPort;
 import fr.insee.genesis.exceptions.GenesisError;
 import fr.insee.genesis.exceptions.GenesisException;
 import fr.insee.genesis.exceptions.SchemaValidationException;
@@ -54,11 +55,13 @@ public class RawResponseController {
     private static final String PARTITION_ID = "partitionId";
     private static final String INTERROGATION_ID = "interrogationId";
     private final LunaticJsonRawDataApiPort lunaticJsonRawDataApiPort;
+    private final RawResponseApiPort rawResponseApiPort;
     private final RawResponseInputRepository rawRepository;
 
 
-    public RawResponseController(LunaticJsonRawDataApiPort lunaticJsonRawDataApiPort, RawResponseInputRepository rawRepository) {
+    public RawResponseController(LunaticJsonRawDataApiPort lunaticJsonRawDataApiPort, RawResponseApiPort rawResponseApiPort, RawResponseInputRepository rawRepository) {
         this.lunaticJsonRawDataApiPort = lunaticJsonRawDataApiPort;
+        this.rawResponseApiPort = rawResponseApiPort;
         this.rawRepository = rawRepository;
     }
 
@@ -186,6 +189,28 @@ public ResponseEntity<String> saveRawResponsesFromRawResponseDto(
         return ResponseEntity.status(201).body(String.format(SUCCESS_MESSAGE, dto.getInterrogationId()));
     }
 
+    //PROCESS
+    @Operation(summary = "Process raw data of a campaign")
+    @PostMapping(path = "/raw-responses/process")
+    @PreAuthorize("hasRole('SCHEDULER')")
+    public ResponseEntity<String> processRawResponses(
+            @RequestParam("questionnaireId") String questionnaireId,
+            @RequestBody List<String> interrogationIdList
+    ) {
+        log.info("Try to process raw responses for questionnaireId {} and {} interrogationIds", interrogationIdList.size());
+        List<GenesisError> errors = new ArrayList<>();
+
+        try {
+            DataProcessResult result = rawResponseApiPort.processRawResponses(questionnaireId, interrogationIdList, errors);
+            return result.formattedDataCount() == 0 ?
+                    ResponseEntity.ok("%d document(s) processed".formatted(result.dataCount()))
+                    : ResponseEntity.ok("%d document(s) processed, including %d FORMATTED after data verification"
+                    .formatted(result.dataCount(), result.formattedDataCount()));
+        } catch (GenesisException e) {
+            return ResponseEntity.status(e.getStatus()).body(e.getMessage());
+        }
+    }
+
     //GET unprocessed
     @Operation(summary = "Get campaign id and interrogationId from all unprocessed raw json data")
     @GetMapping(path = "/responses/raw/lunatic-json/get/unprocessed")

src/main/java/fr/insee/genesis/domain/model/context/DataProcessingContextModel.java

@@ -22,6 +22,8 @@ public class DataProcessingContextModel {
 
     private String partitionId;
 
+    private String collectionInstrumentId; //QuestionnaireId
+
     private LocalDateTime lastExecution;
 
     List<KraftwerkExecutionSchedule> kraftwerkExecutionScheduleList;
@@ -31,6 +33,7 @@ public class DataProcessingContextModel {
     public ScheduleDto toScheduleDto(){
         return ScheduleDto.builder()
                 .surveyName(partitionId)
+                .collectionInstrumentId(collectionInstrumentId)
                 .lastExecution(lastExecution)
                 .kraftwerkExecutionScheduleList(kraftwerkExecutionScheduleList)
                 .build();

src/main/java/fr/insee/genesis/domain/model/surveyunit/SurveyUnitModel.java

@@ -10,12 +10,18 @@
 import java.util.List;
 import java.util.Objects;
 
+/*
+ * This model class should be in equation with the model of our information system ("modèle filière)
+ * Its up to the adapter and mappers to deal with old fields
+ *
+ */
 @Builder
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
 public class SurveyUnitModel {
 
+
 	private String questionnaireId;
 	private String campaignId;
 	private String interrogationId;

src/main/java/fr/insee/genesis/domain/model/surveyunit/rawdata/RawResponse.java

@@ -0,0 +1,18 @@
+package fr.insee.genesis.domain.model.surveyunit.rawdata;
+
+import fr.insee.genesis.domain.model.surveyunit.Mode;
+import org.bson.types.ObjectId;
+
+import java.time.LocalDateTime;
+import java.util.Map;
+
+public record RawResponse(
+        ObjectId id,
+        String interrogationId,
+        String collectionInstrumentId,
+        Mode mode,
+        Map<String,Object> payload,
+        LocalDateTime recordDate,
+        LocalDateTime processDate
+)
+{}

src/main/java/fr/insee/genesis/domain/ports/api/DataProcessingContextApiPort.java

@@ -32,6 +32,9 @@ void saveKraftwerkExecutionSchedule(String partitionId,
 
     DataProcessingContextModel getContext(String interrogationId) throws GenesisException;
     DataProcessingContextModel getContextByPartitionId(String partitionId) throws GenesisException;
+
+    DataProcessingContextModel getContextByCollectionInstrumentId(String collectionInstrumentId);
+
     List<String> getPartitionIds(boolean withReview);
 
     /**

src/main/java/fr/insee/genesis/domain/ports/api/RawResponseApiPort.java

@@ -0,0 +1,19 @@
+package fr.insee.genesis.domain.ports.api;
+
+import fr.insee.bpm.metadata.model.VariablesMap;
+import fr.insee.genesis.domain.model.surveyunit.Mode;
+import fr.insee.genesis.domain.model.surveyunit.SurveyUnitModel;
+import fr.insee.genesis.domain.model.surveyunit.rawdata.DataProcessResult;
+import fr.insee.genesis.domain.model.surveyunit.rawdata.RawResponse;
+import fr.insee.genesis.exceptions.GenesisError;
+import fr.insee.genesis.exceptions.GenesisException;
+
+import java.util.List;
+
+public interface RawResponseApiPort {
+
+    List<RawResponse> getRawResponses(String questionnaireModelId, Mode mode, List<String> interrogationIdList);
+    DataProcessResult processRawResponses(String questionnaireId, List<String> interrogationIdList, List<GenesisError> errors) throws GenesisException;
+    List<SurveyUnitModel> convertRawResponse(List<RawResponse> rawResponses, VariablesMap variablesMap);
+    void updateProcessDates(List<SurveyUnitModel> surveyUnitModels);
+}

src/main/java/fr/insee/genesis/domain/ports/spi/DataProcessingContextPersistancePort.java

@@ -11,6 +11,8 @@ public interface DataProcessingContextPersistancePort {
 
     List<DataProcessingContextModel> findByPartitionIds(List<String> partitionIds);
 
+    List<DataProcessingContextModel> findByCollectionInstrumentIds(List<String> collectionInstrumentIds);
+
     void save(DataProcessingContextDocument dataProcessingContextDocument);
 
     void saveAll(List<DataProcessingContextDocument> dataProcessingContextDocuments);

src/main/java/fr/insee/genesis/domain/ports/spi/RawResponsePersistencePort.java

@@ -0,0 +1,15 @@
+package fr.insee.genesis.domain.ports.spi;
+
+import fr.insee.genesis.domain.model.surveyunit.Mode;
+import fr.insee.genesis.domain.model.surveyunit.rawdata.RawResponse;
+
+import java.util.List;
+import java.util.Set;
+
+public interface RawResponsePersistencePort {
+
+    List<RawResponse> findRawResponses(String questionnaireId, Mode mode, List<String> interrogationIdList);
+
+
+    void updateProcessDates(String collectionInstrumentId, Set<String> interrogationIds);
+}

src/main/java/fr/insee/genesis/domain/service/context/DataProcessingContextService.java

@@ -174,6 +174,13 @@ public DataProcessingContextModel getContextByPartitionId(String partitionId){
         );
     }
 
+    @Override
+    public DataProcessingContextModel getContextByCollectionInstrumentId(String collectionInstrumentId){
+        return DataProcessingContextMapper.INSTANCE.documentToModel(
+                dataProcessingContextPersistancePort.findByPartitionId(collectionInstrumentId)
+        );
+    }
+
     @Override
     public List<String> getPartitionIds(boolean withReview){
         List<String> partitionIds = new ArrayList<>();