Add additional Contention Classification database table columns

db-init/src/main/resources/database/migrations/V1.4__BIE_Contention_Event_Add_additional_Columns.sql

@@ -0,0 +1,15 @@
+ALTER TABLE "bie_contention_event"
+ADD COLUMN "date_added" TIMESTAMP,
+ADD COLUMN "date_completed" TIMESTAMP,
+ADD COLUMN "date_updated" TIMESTAMP,
+ADD COLUMN "actor_station" VARCHAR(255),
+ADD COLUMN "automation_indicator" boolean,
+ADD COLUMN "benefit_claim_type_code" VARCHAR(255),
+ADD COLUMN "contention_status_type_code" VARCHAR(255),
+ADD COLUMN "current_lifecycle_status" VARCHAR(255),
+ADD COLUMN "clmnt_txt" VARCHAR(255),
+ADD COLUMN "details" VARCHAR(255),
+ADD COLUMN "event_time" TIMESTAMP,
+ADD COLUMN "journal_status_type_code" VARCHAR(255),
+ADD COLUMN "veteran_participant_id" VARCHAR(255),
+ADD COLUMN "event_details" JSON;

domain-xample/xample-workflows/src/main/java/gov/va/vro/routes/xample/DbHelper.java

@@ -60,12 +60,27 @@ public ContentionEventEntity saveContentionEvent(final BieMessagePayload bieMess
     contentionEventEntity.setDiagnosticTypeCode(bieMessagePayload.getDiagnosticTypeCode());
     contentionEventEntity.setContentionClassificationName(
         bieMessagePayload.getContentionClassificationName());
-    contentionEventEntity.setOccurredAt(convertTime(bieMessagePayload.getOccurredAt()));
+    contentionEventEntity.setOccurredAt(convertTime(bieMessagePayload.getEventTime()));
+    contentionEventEntity.setDateAdded(convertTime(bieMessagePayload.getDateAdded()));
+    contentionEventEntity.setDateCompleted(convertTime(bieMessagePayload.getDateCompleted()));
+    contentionEventEntity.setDateUpdated(convertTime(bieMessagePayload.getDateUpdated()));
+    contentionEventEntity.setActorStation(bieMessagePayload.getActorStation());
+    contentionEventEntity.setAutomationIndicator(bieMessagePayload.getAutomationIndicator());
+    contentionEventEntity.setBenefitClaimTypeCode(bieMessagePayload.getBenefitClaimTypeCode());
+    contentionEventEntity.setContentionStatusTypeCode(
+        bieMessagePayload.getContentionStatusTypeCode());
+    contentionEventEntity.setCurrentLifecycleStatus(bieMessagePayload.getCurrentLifecycleStatus());
+    contentionEventEntity.setDetails(bieMessagePayload.getDetails());
+    contentionEventEntity.setEventTime(convertTime(bieMessagePayload.getEventTime()));
+    contentionEventEntity.setJournalStatusTypeCode(bieMessagePayload.getJournalStatusTypeCode());
+    contentionEventEntity.setVeteranParticipantId(bieMessagePayload.getVeteranParticipantId());
 
     return contentionEventRepository.save(contentionEventEntity);
   }
 
-  public LocalDateTime convertTime(long time) {
+  public LocalDateTime convertTime(Long time) {
+    if(time == null) return null;
+
     return LocalDateTime.ofInstant(Instant.ofEpochMilli(time), ZoneId.systemDefault());
   }
 }

domain-xample/xample-workflows/src/test/java/gov/va/vro/routes/xample/DbHelperTest.java

@@ -68,7 +68,7 @@ void saveContentionEvent() {
         bieMessagePayload.getContentionClassificationName(),
         entity.getContentionClassificationName());
     assertEquals(bieMessagePayload.getDiagnosticTypeCode(), entity.getDiagnosticTypeCode());
-    assertEquals(dbHelper.convertTime(bieMessagePayload.getOccurredAt()), entity.getOccurredAt());
+    assertEquals(dbHelper.convertTime(bieMessagePayload.getEventTime()), entity.getOccurredAt());
     assertEquals(dbHelper.convertTime(bieMessagePayload.getNotifiedAt()), entity.getNotifiedAt());
   }
 }

scripts/kafka-service.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+export COMPOSE_PROFILES='kafka'
+source scripts/setenv.sh
+
+./gradlew docker
+./gradlew :dockerComposeUp
+./gradlew app:dockerComposeUp
+./gradlew -p mocks docker
+./gradlew -p mocks :dockerComposeUp

scripts/setenv.sh

@@ -199,6 +199,9 @@ export BIP_EVIDENCE_SECRET=daSecret
 #export BIP_APPLICATION_ID=VRO
 export BIP_STATION_ID=456
 
+export KEYSTORE_FILE="keystore.p12"
+export TRUSTSTORE_FILE="truststore.p12"
+
 ###
 ### Integration with BIE's Kafka ###
 ##

shared/lib-bie-kafka/src/main/java/gov/va/vro/model/biekafka/BieMessagePayload.java

@@ -10,16 +10,32 @@
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @EqualsAndHashCode
 public class BieMessagePayload {
+  // these are vro fields
   private Integer status;
   private String statusMessage;
   private ContentionEvent eventType;
-  private long claimId;
-  private long contentionId;
+  private Long notifiedAt;
+
+  // populated from kafka topic payload
+  private String benefitClaimTypeCode;
+  private String actorStation;
+  private String details;
+  private Long veteranParticipantId;
+  private Long claimId;
+  private Long contentionId;
   private String contentionTypeCode;
   private String contentionClassificationName;
   private String diagnosticTypeCode;
   private String actionName;
   private String actionResultName;
-  private long notifiedAt;
-  private long occurredAt;
+  private Boolean automationIndicator;
+  private String contentionStatusTypeCode;
+  private String currentLifecycleStatus;
+  private Long eventTime;
+  private String clmntTxt;
+  private String journalStatusTypeCode;
+  private Long dateAdded;
+  private Long dateCompleted;
+  private Long dateUpdated;
+  private String eventDetails;
 }

shared/lib-bie-kafka/src/main/java/gov/va/vro/model/biekafka/test/BieMessagePayloadFactory.java

@@ -7,22 +7,65 @@
 import java.util.concurrent.TimeUnit;
 
 public class BieMessagePayloadFactory {
-
   private static final Faker faker = new Faker();
 
-  public static BieMessagePayload create() {
+  private static BieMessagePayload createPayload(ContentionEvent eventType) {
     return BieMessagePayload.builder()
-        .eventType(ContentionEvent.CONTENTION_ASSOCIATED_TO_CLAIM)
+        .eventType(eventType)
         .claimId(faker.random().nextLong())
         .contentionClassificationName(faker.lorem().word())
         .contentionTypeCode(faker.lorem().characters(10))
         .contentionId(faker.random().nextLong())
         .diagnosticTypeCode(faker.lorem().characters(10))
-        .occurredAt(faker.date().past(60, TimeUnit.DAYS).getTime())
+        .eventTime(faker.date().past(60, TimeUnit.DAYS).getTime())
         .notifiedAt(faker.date().past(60, TimeUnit.DAYS).getTime())
         .actionName(faker.lorem().characters(10))
         .actionResultName(faker.lorem().characters(10))
         .status(200)
         .build();
   }
+
+  private static void setCommonPayloadValues(BieMessagePayload payload) {
+    payload.setBenefitClaimTypeCode(faker.lorem().characters(10));
+    payload.setActorStation(faker.lorem().characters(200));
+    payload.setDetails(faker.lorem().characters(200));
+    payload.setVeteranParticipantId(faker.random().nextLong());
+    payload.setAutomationIndicator(faker.bool().bool());
+    payload.setContentionStatusTypeCode(faker.lorem().characters(5));
+    payload.setCurrentLifecycleStatus(faker.lorem().characters(8));
+    payload.setEventTime(faker.date().past(60, TimeUnit.DAYS).getTime());
+  }
+
+  private static void setDeletedPayloadValues(BieMessagePayload payload) {
+    payload.setAutomationIndicator(faker.bool().bool());
+    payload.setContentionStatusTypeCode(faker.lorem().characters(5));
+    payload.setCurrentLifecycleStatus(faker.lorem().characters(8));
+    payload.setEventTime(faker.date().past(60, TimeUnit.DAYS).getTime());
+  }
+
+  public static BieMessagePayload create() {
+    // generate a random contention classification event type
+    ContentionEvent eventType = faker.options().option(ContentionEvent.values());
+
+    BieMessagePayload payload = createPayload(eventType);
+
+    switch (eventType) {
+      case CONTENTION_ASSOCIATED_TO_CLAIM:
+      case CONTENTION_CLASSIFIED:
+        setCommonPayloadValues(payload);
+        break;
+      case CONTENTION_UPDATED:
+        setCommonPayloadValues(payload);
+        payload.setJournalStatusTypeCode(faker.lorem().characters(5));
+        break;
+      case CONTENTION_DELETED:
+        setDeletedPayloadValues(payload);
+        break;
+      case CONTENTION_COMPLETED:
+        // No special handling for CONTENTION_COMPLETED
+        break;
+    }
+
+    return payload;
+  }
 }

shared/persistence-model/src/main/java/gov/va/vro/persistence/model/ContentionEventEntity.java

@@ -37,4 +37,43 @@ public class ContentionEventEntity extends BaseEntity {
 
   @Column(name = "diagnostic_type_code")
   private String diagnosticTypeCode;
+
+  @Column(name = "date_added")
+  private LocalDateTime dateAdded;
+
+  @Column(name = "date_completed")
+  private LocalDateTime dateCompleted;
+
+  @Column(name = "date_updated")
+  private LocalDateTime dateUpdated;
+
+  @Column(name = "actor_station")
+  private String actorStation;
+
+  @Column(name = "automation_indicator")
+  private boolean automationIndicator;
+
+  @Column(name = "benefit_claim_type_code")
+  private String benefitClaimTypeCode;
+
+  @Column(name = "contention_status_type_code")
+  private String contentionStatusTypeCode;
+
+  @Column(name = "current_lifecycle_status")
+  private String currentLifecycleStatus;
+
+  @Column(name = "details")
+  private String details;
+
+  @Column(name = "event_time")
+  private LocalDateTime eventTime;
+
+  @Column(name = "journal_status_type_code")
+  private String journalStatusTypeCode;
+
+  @Column(name = "veteran_participant_id")
+  private long veteranParticipantId;
+
+  @Column(name = "event_details")
+  private String eventDetails;
 }

svc-bie-kafka/src/integrationTest/java/gov/va/vro/services/bie/BieKafkaApplicationTest.java

@@ -67,7 +67,7 @@ public void sendEventToKafkaTopic() throws InterruptedException, IOException {
     // Message 2 comes through Kafka
     BieMessagePayload kafkaEventBody = BieMessagePayloadFactory.create();
     kafkaEventBody.setEventType(null);
-    kafkaEventBody.setContentionId(1234567890);
+    kafkaEventBody.setContentionId(1234567890L);
 
     ObjectMapper objectMapper = new ObjectMapper();
     val kafkaSentMessage = objectMapper.writeValueAsString(kafkaEventBody);

svc-bie-kafka/src/main/java/gov/va/vro/services/bie/service/kafka/KafkaConsumer.java

@@ -6,6 +6,7 @@
 import gov.va.vro.model.biekafka.ContentionEvent;
 import gov.va.vro.services.bie.config.BieProperties;
 import gov.va.vro.services.bie.service.AmqpMessageSender;
+import gov.va.vro.services.bie.utils.BieMessageUtils;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.avro.generic.GenericRecord;
@@ -31,6 +32,7 @@ public void consume(ConsumerRecord<String, Object> record) {
       log.info("Topic name: {}", topicName);
       if (record.value() instanceof GenericRecord) {
         payload = this.handleGenericRecord(record);
+        log.info("kafka payload: {}", payload);
       } else if (record.value() instanceof String stringPayload) {
         log.info("Consumed message string value (before) json conversion: {}", stringPayload);
         payload = this.handleStringRecord(record);
@@ -56,28 +58,12 @@ private BieMessagePayload handleStringRecord(ConsumerRecord<String, Object> reco
 
   private BieMessagePayload handleGenericRecord(ConsumerRecord<String, Object> record) {
     GenericRecord messageValue = (GenericRecord) record.value();
-    String KEY_CONTENTION_CLASSIFICATION_NAME = "ContentionClassificationName";
-    String KEY_DIAGNOSTIC_TYPE_CODE = "DiagnosticTypeCode";
-    String KEY_CLAIM_ID = "ClaimId";
-    String KEY_CONTENTION_ID = "ContentionId";
-    String KEY_CONTENTION_TYPE_CODE = "ContentionTypeCode";
-    String KEY_EVENT_TIME = "EventTime";
-    String ACTION_NAME = "ActionName";
-    String ACTION_RESULT_NAME = "ActionResultName";
 
-    return BieMessagePayload.builder()
-        .eventType(
-            ContentionEvent.valueOf(ContentionEvent.mapTopicToEvent(record.topic()).toString()))
-        .claimId((long) messageValue.get(KEY_CLAIM_ID))
-        .contentionId((long) messageValue.get(KEY_CONTENTION_ID))
-        .contentionClassificationName((String) messageValue.get(KEY_CONTENTION_CLASSIFICATION_NAME))
-        .contentionTypeCode((String) messageValue.get(KEY_CONTENTION_TYPE_CODE))
-        .diagnosticTypeCode((String) messageValue.get(KEY_DIAGNOSTIC_TYPE_CODE))
-        .occurredAt((Long) messageValue.get(KEY_EVENT_TIME))
-        .notifiedAt(record.timestamp())
-        .actionName((String) messageValue.get(ACTION_NAME))
-        .actionResultName((String) messageValue.get(ACTION_RESULT_NAME))
-        .status(200)
-        .build();
+    BieMessagePayload payload = BieMessageUtils.processBieMessagePayloadFields(messageValue);
+    payload.setEventType(
+        ContentionEvent.valueOf(ContentionEvent.mapTopicToEvent(record.topic()).toString()));
+    payload.setNotifiedAt(record.timestamp());
+
+    return payload;
   }
 }

svc-bie-kafka/src/main/java/gov/va/vro/services/bie/utils/BieMessageUtils.java

@@ -0,0 +1,63 @@
+package gov.va.vro.services.bie.utils;
+
+import gov.va.vro.model.biekafka.BieMessagePayload;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.avro.generic.GenericRecord;
+import org.springframework.util.StringUtils;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.HashSet;
+import java.util.Set;
+
+@Slf4j
+public class BieMessageUtils {
+  private static final Set<String> IGNORED_FIELDS = new HashSet<>();
+
+  static {
+    // Add the names of fields to be ignored
+    IGNORED_FIELDS.add("status");
+    IGNORED_FIELDS.add("statusMessage");
+    IGNORED_FIELDS.add("eventType");
+    IGNORED_FIELDS.add("notifiedAt");
+  }
+
+  public static BieMessagePayload processBieMessagePayloadFields(GenericRecord genericRecord) {
+    BieMessagePayload payload = BieMessagePayload.builder().status(200).build();
+
+    for (Field field : BieMessagePayload.class.getDeclaredFields()) {
+      String fieldName = field.getName();
+
+      // Skip the field if it's in the ignored list
+      if (IGNORED_FIELDS.contains(fieldName)) {
+        continue;
+      }
+
+      String capitalizedFieldName = StringUtils.capitalize(fieldName);
+      Object value = genericRecord.get(capitalizedFieldName);
+
+      if (value == null) {
+        continue; // Skip setting the field if the value is null
+      }
+
+      try {
+        String setterMethodName = "set" + capitalizedFieldName;
+        Method setterMethod = BieMessagePayload.class.getMethod(setterMethodName, field.getType());
+
+        if (field.getType().isAssignableFrom(value.getClass())) {
+          setterMethod.invoke(payload, value);
+        } else {
+          log.warn(
+              "Type mismatch for field '{}'. Expected type: '{}', Actual value: '{}'",
+              fieldName,
+              field.getType().getSimpleName(),
+              value);
+        }
+      } catch (Exception e) {
+        log.error("Error setting value for field '{}': {}", fieldName, e.getMessage(), e);
+      }
+    }
+
+    return payload;
+  }
+}

svc-bie-kafka/src/main/resources/application-dev.yaml

@@ -11,7 +11,7 @@ spring:
       schema.registry.ssl.truststore.password: "${BIE_KAFKA_TRUSTSTORE_PASSWORD}"
       schema.registry.ssl.truststore.type: "PKCS12"
     consumer:
-      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:vro-bie-tst-vro-2}"
+      group-id: "${BIE_KAFKA_PLACEHOLDERS_GROUP_ID:vro-bie-tst-vro-3}"
       key-deserializer: "org.apache.kafka.common.serialization.StringDeserializer"
       value-deserializer: "io.confluent.kafka.serializers.KafkaAvroDeserializer"
       properties: