Merge branch 'dev' into CU-86c0dgwg9_API-query-endpoint

Author: MatthewErispe

JeMPI_Apps/JeMPI_AsyncReceiver/src/main/java/org/jembi/jempi/async_receiver/Main.java

@@ -15,6 +15,7 @@
 import org.jembi.jempi.shared.serdes.JsonPojoSerializer;
 import org.jembi.jempi.shared.utils.AppUtils;
 import org.apache.commons.codec.language.Soundex;
+import org.jetbrains.annotations.NotNull;
 
 import java.io.IOException;
 import java.nio.file.*;
@@ -153,28 +154,7 @@ private void apacheReadCSV(
 
             for (CSVRecord csvRecord : csvParser) {
                final var patientRecord = demographicData(csvRecord);
-               String givenName = patientRecord.fields.stream()
-                  .filter(field -> "given_name".equals(field.ccTag()))
-                  .map(DemographicData.DemographicField::value)
-                  .findFirst()
-                  .orElse("");
-               String familyName = patientRecord.fields.stream()
-                  .filter(field -> "family_name".equals(field.ccTag()))
-                  .map(DemographicData.DemographicField::value)
-                  .findFirst()
-                  .orElse("");
-
-               String partitionKey = "";
-               if (!givenName.isEmpty()) {
-                  partitionKey += new Soundex().soundex(givenName);
-               }
-               if (!familyName.isEmpty()) {
-                  partitionKey += new Soundex().soundex(familyName);
-               }
-               if (givenName.isEmpty() && familyName.isEmpty()) {
-                  partitionKey += "Unknown";
-               }
-               LOGGER.info("Using Kafka topic/partition: " + partitionKey);
+               String partitionKey = generateKafkaPartitionKey(patientRecord);
 
                final var interactionEnvelop = new InteractionEnvelop(InteractionEnvelop.ContentType.BATCH_INTERACTION,
                                                                      tag,
@@ -203,6 +183,30 @@ private void apacheReadCSV(
       }
    }
 
+   private static @NotNull String generateKafkaPartitionKey(final DemographicData patientRecord) {
+      StringBuilder partitionKey = new StringBuilder();
+
+      for (String field : FIELDS_CONFIG.fieldsForKafkaKeyGen) {
+         String value = patientRecord.fields.stream()
+                                            .filter(fieldData -> field.equals(fieldData.ccTag()))
+                                            .map(DemographicData.DemographicField::value)
+                                            .findFirst()
+                                            .orElse("");
+
+         if (!value.isEmpty()) {
+            partitionKey.append(new Soundex().soundex(value));
+         }
+      }
+
+      if (partitionKey.length() == 0) {
+         partitionKey.append("Unknown");
+      }
+
+      String partitionKeyStr = partitionKey.toString();
+      LOGGER.info("Using Kafka topic/partition: {}", partitionKeyStr);
+      return partitionKeyStr;
+   }
+
    private String updateStan(
          final String stanDate,
          final int recCount) {

JeMPI_Apps/JeMPI_EM_Scala/src/main/scala/org/jembi/jempi/em/config/Config.scala

@@ -61,9 +61,11 @@ case class Rules(
 )
 
 case class Config(
-    auxInteractionFields: Option[Array[auxField]],
-    auxGoldenRecordFields: Option[Array[auxField]],
-    additionalNodes: Option[Array[AdditionalNode]],
-    demographicFields: Array[DemographicField],
-    rules: Rules
+                   auxInteractionFields: Option[Array[auxField]],
+                   auxGoldenRecordFields: Option[Array[auxField]],
+                   additionalNodes: Option[Array[AdditionalNode]],
+                   demographicFields: Array[DemographicField],
+                   rules: Rules,
+                   fieldsForKafkaKeyGen: Option[List[String]],
+                   nameFieldsForNotificationDisplay: Option[List[String]]
 )

JeMPI_Apps/JeMPI_LibShared/src/main/java/org/jembi/jempi/shared/config/FieldsConfig.java

@@ -31,6 +31,8 @@ public final class FieldsConfig {
    public final List<AuxField> userAuxInteractionFields;
    public final List<AuxField> userAuxGoldenRecordFields;
    public final List<DemographicField> demographicFields;
+   public final List<String> fieldsForKafkaKeyGen;
+   public final List<String> nameFieldsForNotificationDisplay;
    public final List<AdditionalNode> additionalNodes;
 
    public FieldsConfig(final JsonConfig jsonConfig) {
@@ -88,6 +90,13 @@ public FieldsConfig(final JsonConfig jsonConfig) {
                        );
       optionalInteractionAuxIdIdx = auxInteractionAuxIdIdx[0];
 
+      fieldsForKafkaKeyGen = jsonConfig.fieldsForKafkaKeyGen() != null
+            ? List.copyOf(jsonConfig.fieldsForKafkaKeyGen())
+            : List.of();
+
+      nameFieldsForNotificationDisplay = jsonConfig.nameFieldsForNotificationDisplay() != null
+            ? List.copyOf(jsonConfig.nameFieldsForNotificationDisplay())
+            : List.of();
 
       demographicFields = jsonConfig.demographicFields()
                                     .stream()

JeMPI_Apps/JeMPI_LibShared/src/main/java/org/jembi/jempi/shared/config/input/JsonConfig.java

@@ -19,6 +19,8 @@ public record JsonConfig(
       @JsonProperty("auxGoldenRecordFields") List<AuxGoldenRecordField> auxGoldenRecordFields,
       List<AdditionalNode> additionalNodes,
       List<DemographicField> demographicFields,
+      @JsonProperty("fieldsForKafkaKeyGen") List<String> fieldsForKafkaKeyGen,
+      @JsonProperty("nameFieldsForNotificationDisplay") List<String> nameFieldsForNotificationDisplay,
       Rules rules) {
 
    private static final Logger LOGGER = LogManager.getLogger(JsonConfig.class);

JeMPI_Apps/JeMPI_LibShared/src/main/java/org/jembi/jempi/shared/config/input/NameFieldsForKafkaKeyGen.java

@@ -0,0 +1,10 @@
+package org.jembi.jempi.shared.config.input;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+import java.util.List;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+   public record NameFieldsForKafkaKeyGen(
+         List<String> nameFieldsForKafkaKeyGen
+         ) {
+   }

JeMPI_Apps/JeMPI_LibShared/src/main/java/org/jembi/jempi/shared/models/ConfigurationModel.java

@@ -16,6 +16,8 @@ public record Configuration(
                         List<Field> auxGoldenRecordFields,
                         List<Node> additionalNodes,
                         List<DemographicField> demographicFields,
+                        List<String> fieldsForKafkaKeyGen,
+                        List<String> nameFieldsForNotificationDisplay,
                         Rules rules) {
         }
 

JeMPI_Apps/JeMPI_Linker/src/main/java/org/jembi/jempi/linker/backend/LinkerDWH.java

@@ -95,17 +95,17 @@ private static boolean helperUpdateGoldenRecordField(
 
    private static String patientName(final Interaction interaction) {
       var patientRecord = interaction.demographicData();
-      String givenName = patientRecord.fields.stream()
-         .filter(field -> "given_name".equals(field.ccTag()))
-         .map(DemographicData.DemographicField::value)
-         .findFirst()
-         .orElse("");
-      String familyName = patientRecord.fields.stream()
-         .filter(field -> "family_name".equals(field.ccTag()))
-         .map(DemographicData.DemographicField::value)
-         .findFirst()
-         .orElse("");
-         return (givenName + " " + familyName).trim();
+      Map<String, String> fieldMap = patientRecord.fields.stream()
+                                                         .collect(Collectors.toMap(DemographicData.DemographicField::ccTag,
+                                                                                   DemographicData.DemographicField::value));
+
+      String patientDisplayName = FIELDS_CONFIG.nameFieldsForNotificationDisplay.stream()
+                                                                                .map(fieldName -> fieldMap.getOrDefault(fieldName,
+                                                                                                                        ""))
+                                                                                .filter(StringUtils::isNotBlank)
+                                                                                .collect(Collectors.joining(" "))
+                                                                                .trim();
+      return patientDisplayName;
    }
 
    /**

devops/linux/docker/backup_restore/dgraph-restore-api.py

@@ -9,6 +9,47 @@
 
 host = env_vars['NODE1_IP']
 port = "50010"
+endpoint = "http://" + host + ":6080"
+new_lease_value = 50000000000000
+
+def get_current_lease(endpoint):
+    """Fetches the current maxLeasedUid from the Dgraph Zero /state endpoint."""
+    try:
+        response = requests.get(f"{endpoint}/state")
+        response.raise_for_status()  # Raise an exception for HTTP errors
+        data = response.json()
+        # Extracting the maxLeasedUid from the JSON response
+        max_leased_uid = data.get('maxUID', None)
+        if max_leased_uid is not None:
+            print(f"Current maxLeasedUid: {max_leased_uid}")
+            return max_leased_uid
+        else:
+            print("Error: maxLeasedUid not found in the response.")
+            return None
+    except requests.RequestException as e:
+        print(f"Error fetching current lease: {e}")
+        return None
+
+def increase_lease(endpoint, new_value):
+    """Increases the lease value if it's below the new_value."""
+    current_value = get_current_lease(endpoint)
+    current_value = int(current_value)
+    if current_value is None:
+        print("Unable to fetch the current lease value. Exiting.")
+        return
+
+    if current_value < new_value:
+        try:
+            # Assuming a POST request with form data to increase lease size
+            data = {'what': 'uids', 'num': new_value}
+            response = requests.get(f"{endpoint}/assign", params=data)
+            response.raise_for_status()  # Raise an exception for HTTP errors
+            print(f"Lease increased to {new_value}.")
+        except requests.RequestException as e:
+            print(f"Error increasing lease: {e}")
+            print(f"Response Content: {response.content}")  # Debug print to see response content
+    else:
+        print(f"Current lease ({current_value}) is already greater than or equal to {new_value}. No action taken.")
 
 
 def main(json_file):
@@ -54,16 +95,15 @@ def convert_datetime_format(date_str):
             continue
     else:
         return date_str  # If the format is not correct, return the original string
-    
+
     output_format = "%Y-%m-%dT%H:%M:%S.%fZ"
     output_str = dt.strftime(output_format)
     output_str = output_str[:26] + 'Z'  # Keep only the first 2 decimal places of the seconds part
     return output_str
 
 def process_json_data(golden_records):
-    
+    increase_lease(endpoint, new_lease_value)
     for golden_record in golden_records:
-        
         golden_record['goldenRecord']['uniqueGoldenRecordData']['auxDateCreated'] = convert_datetime_format(golden_record['goldenRecord']['uniqueGoldenRecordData']['auxDateCreated'])
         for interaction in golden_record['interactionsWithScore']:
             interaction['interaction']['uniqueInteractionData']['auxDateCreated'] = convert_datetime_format(
@@ -73,7 +113,7 @@ def process_json_data(golden_records):
         response = send_golden_record_to_api(golden_record)
         if response:
             print("After Restore Golden ID--"+ response.text)
-    
+
 def send_golden_record_to_api(golden_record_payload):
     get_expanded_golden_record_url = f'http://{host}:{port}/JeMPI/restoreGoldenRecord'
     # Normalize date fields in the payload

devops/linux/docker/data-config/config-reference-auto-generate-fields.json

@@ -135,6 +135,14 @@
       "indexGoldenRecord": "@index(exact)"
     }
   ],
+  "fieldsForKafkaKeyGen": [
+    "given_name",
+    "family_name"
+  ],
+  "nameFieldsForNotificationDisplay": [
+    "given_name",
+    "family_name"
+  ],
   "rules": {
     "link": {
       "deterministic": [

devops/linux/docker/data-config/config-reference-link-d-validate-dp-match-dp.json

@@ -222,6 +222,14 @@
       "matchMetaData": null
     }
   ],
+  "fieldsForKafkaKeyGen": [
+    "given_name",
+    "family_name"
+  ],
+  "nameFieldsForNotificationDisplay": [
+    "given_name",
+    "family_name"
+  ],
   "rules": {
     "link": {
       "deterministic": [