Merge pull request #26 from pagopa/PAGOPA-2060-recovery-silandro-flows

BUG: [PAGOPA-2060] - silandro flows

.devops/code-review-pipelines.yml

@@ -36,8 +36,8 @@ steps:
       extraProperties: |
         sonar.projectKey=$(SONARCLOUD_PROJECT_KEY)
         sonar.projectName=$(SONARCLOUD_PROJECT_NAME)
-        sonar.coverage.exclusions=**/config/*,**/*Mock*,**/model/**,**/entity/*
-        sonar.cpd.exclusions=**/model/**,**/entity/*
+        sonar.coverage.exclusions=**/config/*,**/*Mock*,**/models/**,**/entity/*,**/client/*,**/exception/* 
+        sonar.cpd.exclusions=**/models/**,**/entity/*
       
 
   - task: Maven@3

pom.xml

@@ -19,6 +19,7 @@
         <azure.storage.version>8.6.6</azure.storage.version>
         <resteasy.version>3.15.1.Final</resteasy.version>
         <functionAppName>reporting-service-20210712115055634</functionAppName>
+        <google-api-client.version>2.7.0</google-api-client.version>
     </properties>
 
     <dependencies>
@@ -115,6 +116,12 @@
             <artifactId>lombok</artifactId>
             <version>1.18.20</version>
         </dependency>
+
+        <dependency>
+            <groupId>com.google.api-client</groupId>
+            <artifactId>google-api-client-gson</artifactId>
+            <version>${google-api-client.version}</version>
+        </dependency>
 
         <!-- Util END-->
 

src/main/java/it/gov/pagopa/reporting/client/ApiConfigClient.java

@@ -0,0 +1,129 @@
+package it.gov.pagopa.reporting.client;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.api.client.http.GenericUrl;
+import com.google.api.client.http.HttpBackOffUnsuccessfulResponseHandler;
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpRequest;
+import com.google.api.client.http.HttpRequestFactory;
+import com.google.api.client.http.HttpResponseException;
+import com.google.api.client.http.HttpTransport;
+import com.google.api.client.http.javanet.NetHttpTransport;
+import com.google.api.client.json.JsonFactory;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.client.util.ExponentialBackOff;
+
+import it.gov.pagopa.reporting.exception.Cache4XXException;
+import it.gov.pagopa.reporting.exception.Cache5XXException;
+import it.gov.pagopa.reporting.models.cache.CacheResponse;
+import it.gov.pagopa.reporting.models.cache.CreditorInstitutionStation;
+import it.gov.pagopa.reporting.models.cache.Station;
+
+public class ApiConfigClient {
+
+    private static ApiConfigClient instance = null;
+
+    private final HttpTransport httpTransport = new NetHttpTransport();
+    private final JsonFactory jsonFactory = new GsonFactory();
+    private final String apiConfigCacheHost = System.getenv().getOrDefault("CACHE_CLIENT_HOST", ""); // es: https://api.xxx.platform.pagopa.it
+    private final String getCacheDetails = System.getenv().getOrDefault("CACHE_PATH", "/cache?keys=creditorInstitutionStations,stations");
+    private final String apiKey = System.getenv().getOrDefault("CACHE_API_KEY", "");
+
+    // Retry ExponentialBackOff config
+    private final boolean enableRetry        = Boolean.parseBoolean(System.getenv().getOrDefault("ENABLE_CLIENT_RETRY", "false"));
+    private final int initialIntervalMillis  = Integer.parseInt(System.getenv().getOrDefault("INITIAL_INTERVAL_MILLIS", "500"));
+    private final int maxElapsedTimeMillis   = Integer.parseInt(System.getenv().getOrDefault("MAX_ELAPSED_TIME_MILLIS", "1000"));
+    private final int maxIntervalMillis      = Integer.parseInt(System.getenv().getOrDefault("MAX_INTERVAL_MILLIS", "1000"));
+    private final double multiplier          = Double.parseDouble(System.getenv().getOrDefault("MULTIPLIER", "1.5"));
+    private final double randomizationFactor = Double.parseDouble(System.getenv().getOrDefault("RANDOMIZATION_FACTOR", "0.5"));
+
+    public static ApiConfigClient getInstance() {
+        if (instance == null) {
+            instance = new ApiConfigClient();
+        }
+        return instance;
+    }
+
+    public CacheResponse getCache() throws IOException, IllegalArgumentException, Cache5XXException, Cache4XXException {
+        GenericUrl url = new GenericUrl(apiConfigCacheHost + getCacheDetails);
+        HttpRequest request = this.buildGetRequestToApiConfigCache(url);
+
+        if (enableRetry) {
+            this.setRequestRetry(request);
+        }
+
+        return this.executeCallToApiConfigCache(request);
+    }
+
+    public HttpRequest buildGetRequestToApiConfigCache(GenericUrl url) throws IOException {
+
+        HttpRequestFactory requestFactory = httpTransport.createRequestFactory(
+                (HttpRequest request) ->
+                        request.setParser(new JsonObjectParser(jsonFactory))
+        );
+
+        HttpRequest request = requestFactory.buildGetRequest(url);
+        HttpHeaders headers = request.getHeaders();
+        headers.set("Ocp-Apim-Subscription-Key", apiKey);
+        return request;
+    }
+
+    public void setRequestRetry(HttpRequest request) {
+        /**
+         * Retry section config
+         */
+        ExponentialBackOff backoff = new ExponentialBackOff.Builder()
+                .setInitialIntervalMillis(initialIntervalMillis)
+                .setMaxElapsedTimeMillis(maxElapsedTimeMillis)
+                .setMaxIntervalMillis(maxIntervalMillis)
+                .setMultiplier(multiplier)
+                .setRandomizationFactor(randomizationFactor)
+                .build();
+
+        // Exponential Backoff is turned off by default in HttpRequest -> it's necessary include an instance of HttpUnsuccessfulResponseHandler to the HttpRequest to activate it
+        // The default back-off on anabnormal HTTP response is BackOffRequired.ON_SERVER_ERROR (5xx)
+        request.setUnsuccessfulResponseHandler(
+                new HttpBackOffUnsuccessfulResponseHandler(backoff));
+    }
+
+    public CacheResponse executeCallToApiConfigCache(HttpRequest request) throws IOException, IllegalArgumentException, Cache5XXException, Cache4XXException {
+
+        ObjectMapper mapper = new ObjectMapper();
+        CacheResponse cacheResponse = CacheResponse.builder().build();
+        List<CreditorInstitutionStation> creditorInstitutionStationList = new ArrayList<>();
+        List<Station> stationList = new ArrayList<>();
+        try {
+            InputStream resIs = request.execute().getContent();
+            Map<String,Object> responseMap = mapper.readValue(resIs, HashMap.class);
+            Map<String,Object> creditorInstitutionStations = (HashMap) responseMap.get("creditorInstitutionStations");
+            for (Map.Entry<String, Object> creditorInstitutionStation : creditorInstitutionStations.entrySet()) {
+                creditorInstitutionStationList.add(mapper.readValue(mapper.writeValueAsString(creditorInstitutionStation.getValue()), CreditorInstitutionStation.class));
+            }
+            Map<String,Object> stations = (HashMap) responseMap.get("stations");
+            for (Map.Entry<String, Object> station : stations.entrySet()) {
+                stationList.add(mapper.readValue(mapper.writeValueAsString(station.getValue()), Station.class));
+            }
+            cacheResponse.setStations(stationList);
+            cacheResponse.setCreditorInstitutionStations(creditorInstitutionStationList);
+        } catch (HttpResponseException e) {
+            if (e.getStatusCode() / 100 == 4) {
+                String message = String.format("Error %s calling the service URL %s", e.getStatusCode(), request.getUrl());
+                throw new Cache4XXException(message);
+
+            } else if (e.getStatusCode() / 100 == 5) {
+                String message = String.format("Error %s calling the service URL %s", e.getStatusCode(), request.getUrl());
+                throw new Cache5XXException(message);
+
+            }
+        }
+        return cacheResponse;
+    }
+}

src/main/java/it/gov/pagopa/reporting/exception/AppException.java

@@ -0,0 +1,18 @@
+package it.gov.pagopa.reporting.exception;
+
+public class AppException extends Exception {
+
+	/**
+	 * generated serialVersionUID
+	 */
+	private static final long serialVersionUID = -7564079264281462536L;
+
+	public AppException() {
+		super();
+	}
+
+	public AppException(String message) {
+		super(message);
+	}
+
+}

src/main/java/it/gov/pagopa/reporting/exception/Cache4XXException.java

@@ -0,0 +1,18 @@
+package it.gov.pagopa.reporting.exception;
+
+public class Cache4XXException extends Exception {
+
+	/**
+	 * generated serialVersionUID
+	 */
+	private static final long serialVersionUID = -7564079264281462536L;
+
+	public Cache4XXException() {
+		super();
+	}
+
+	public Cache4XXException(String message) {
+		super(message);
+	}
+
+}

src/main/java/it/gov/pagopa/reporting/exception/Cache5XXException.java

@@ -0,0 +1,18 @@
+package it.gov.pagopa.reporting.exception;
+
+public class Cache5XXException extends Exception {
+
+	/**
+	 * generated serialVersionUID
+	 */
+	private static final long serialVersionUID = -7564079264281462536L;
+
+	public Cache5XXException() {
+		super();
+	}
+
+	public Cache5XXException(String message) {
+		super(message);
+	}
+
+}

src/main/java/it/gov/pagopa/reporting/functions/RetrieveDetails.java

@@ -5,17 +5,33 @@
 import com.microsoft.azure.functions.ExecutionContext;
 import com.microsoft.azure.functions.annotation.FunctionName;
 import com.microsoft.azure.functions.annotation.QueueTrigger;
+
+import it.gov.pagopa.reporting.client.ApiConfigClient;
+import it.gov.pagopa.reporting.exception.AppException;
 import it.gov.pagopa.reporting.models.FlowsMessage;
+import it.gov.pagopa.reporting.models.cache.CacheResponse;
+import it.gov.pagopa.reporting.models.cache.CreditorInstitutionStation;
+import it.gov.pagopa.reporting.models.cache.Station;
 import it.gov.pagopa.reporting.service.FlowsService;
 
+import java.time.LocalDate;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 /**
  * FlowsXmlDownloadFunction Azure Functions with Azure Queue trigger.
  */
 public class RetrieveDetails {
+
+	private static CacheResponse cacheContent;
+	private static final Lock cacheLock = new ReentrantLock();
+
     /**
      * This function will be invoked when a new message is detected in the queue
      * FLOWS_QUEUE related to FLOW_SA_CONNECTION_STRING (app settings)
@@ -34,7 +50,7 @@ public void run(
             FlowsMessage flows = new ObjectMapper().readValue(message, FlowsMessage.class);
 
             // retrieve fdr from node
-            this.getFlowsServiceInstance(logger)
+            this.getFlowsServiceInstance(logger, flows.getIdPA())
                     .flowsXmlDownloading(Arrays.asList(flows.getFlows()), flows.getIdPA(), flows.getRetry() + 1);
 
             logger.log(Level.INFO, () -> "[FlowsDownloadFunction END]  processed a message " + message);
@@ -49,17 +65,76 @@ public void run(
         }
     }
 
-    public FlowsService getFlowsServiceInstance(Logger logger) {
+    public FlowsService getFlowsServiceInstance(Logger logger, String idPA) throws AppException {
         String maxRetryQueuing = getVars("MAX_RETRY_QUEUING");
         String queueRetentionSec = getVars("QUEUE_RETENTION_SEC");
         String queueDelaySec = getVars("QUEUE_DELAY_SEC");
-        return new FlowsService(System.getenv("FLOW_SA_CONNECTION_STRING"), System.getenv("PAA_ID_INTERMEDIARIO"),
-                System.getenv("PAA_STAZIONE_INT"), System.getenv("PAA_PASSWORD"), System.getenv("FLOWS_XML_BLOB"), System.getenv("FLOWS_QUEUE"),
+
+        ApiConfigClient cacheClient = this.getCacheClientInstance();
+        // Check if cache update is needed and attempt to acquire lock
+        if (isCacheUpdateNeeded() && (cacheLock.tryLock())) {
+                try {
+                    // Double-check if cache still needs updating within the lock.
+                    if (isCacheUpdateNeeded()) {
+                    	RetrieveDetails.setCache(cacheClient);
+                    }
+                } finally {
+                    cacheLock.unlock();
+                }
+
+        }
+
+        logger.log(Level.INFO, () -> "[RetrieveDetails][Config-Cache][Start] idPa: " + idPA);
+        Station stationBroker = getPAStationIntermediario(idPA)
+                .orElseThrow(() -> new AppException(String.format("No data present in api config database for PA %s", idPA)));
+        String idBroker = stationBroker.getBrokerCode();
+        String idStation = stationBroker.getStationCode();
+        String stationPassword = stationBroker.getPassword();
+
+        return new FlowsService(System.getenv("FLOW_SA_CONNECTION_STRING"), idBroker,
+        		idStation, stationPassword, System.getenv("FLOWS_XML_BLOB"), System.getenv("FLOWS_QUEUE"),
                 System.getenv("FLOWS_TABLE"), Integer.parseInt(maxRetryQueuing), Integer.parseInt(queueRetentionSec), Integer.parseInt(queueDelaySec),
                 logger);
     }
 
     public String getVars(String vars) {
         return System.getenv(vars);
     }
+
+    public ApiConfigClient getCacheClientInstance() {
+        return ApiConfigClient.getInstance();
+    }
+
+    public static void setCacheContent(CacheResponse cacheContent) {
+		RetrieveDetails.cacheContent = cacheContent;
+	}
+
+    private boolean isCacheUpdateNeeded() {
+        return cacheContent == null || (cacheContent.getRetrieveDate() != null &&
+                cacheContent.getRetrieveDate().isBefore(LocalDate.now()));
+    }
+
+    private static void setCache(ApiConfigClient cacheClient) throws AppException {
+        try {
+            cacheContent = cacheClient.getCache();
+            cacheContent.setRetrieveDate(LocalDate.now());
+        } catch (Exception e) {
+            cacheContent = null;
+            throw new AppException(e.getMessage());
+        }
+    }
+
+    private Optional<Station> getPAStationIntermediario(String idPa) {
+        List<String> stationPa = getStations(idPa);
+        return cacheContent.getStations().stream()
+                .filter(station -> stationPa.contains(station.getStationCode()))
+                .filter(Station::getEnabled)
+                .findFirst();
+    }
+
+    private List<String> getStations(String idPa) {
+        return cacheContent.getCreditorInstitutionStations().stream()
+                .filter(creditorInstitutionStation -> creditorInstitutionStation.getCreditorInstitutionCode().equals(idPa))
+                .map(CreditorInstitutionStation::getStationCode).collect(Collectors.toList());
+    }
 }

src/main/java/it/gov/pagopa/reporting/models/cache/CacheResponse.java

@@ -0,0 +1,27 @@
+package it.gov.pagopa.reporting.models.cache;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.*;
+
+import java.time.LocalDate;
+import java.util.List;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Getter
+@Setter
+@ToString
+@JsonIgnoreProperties(ignoreUnknown = true)
+@Builder
+public class CacheResponse {
+
+    @JsonProperty(value = "stations")
+    private List<Station> stations;
+
+    @JsonProperty(value = "creditorInstitutionStations")
+    private List<CreditorInstitutionStation> creditorInstitutionStations;
+
+    private LocalDate retrieveDate;
+
+}

src/main/java/it/gov/pagopa/reporting/models/cache/CreditorInstitutionStation.java

@@ -0,0 +1,22 @@
+package it.gov.pagopa.reporting.models.cache;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.*;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Getter
+@Setter
+@ToString
+@JsonIgnoreProperties(ignoreUnknown = true)
+@Builder
+public class CreditorInstitutionStation {
+
+    @JsonProperty(value = "creditor_institution_code")
+    private String creditorInstitutionCode;
+
+    @JsonProperty(value = "station_code")
+    private String stationCode;
+}