Merge pull request #68

Added authority flow for independents

.github/workflows/deploy-to.openshift-dev.yml

@@ -7,6 +7,8 @@ env:
   OPENSHIFT_TOKEN: ${{ secrets.OPENSHIFT_TOKEN }}
   OPENSHIFT_NAMESPACE_DEV: ${{ secrets.APP_NAMESPACE_NO_ENV }}-dev
 
+  INDEPENDENT_SCHOOLS_NAMESPACE_DEV: ${{ secrets.INDEPENDENT_SCHOOLS_NAMESPACE_NO_ENV }}-dev
+
   DB_JDBC_CONNECT_STRING: ${{ secrets.DB_JDBC_CONNECT_STRING }}
   DB_PWD: ${{ secrets.DB_PWD }}
   DB_USER: ${{ secrets.DB_USER }}
@@ -160,8 +162,7 @@ jobs:
         oc process -f tools/openshift/api.dc.yaml -p APP_NAME=${{ env.APP_NAME }} -p REPO_NAME=${{ env.REPO_NAME }} -p BRANCH=${{ env.BRANCH }} -p NAMESPACE=${{ env.OPENSHIFT_NAMESPACE_DEV }} -p TAG=${{ env.TAG }} -p MIN_REPLICAS=${{ env.MIN_REPLICAS_DEV }} -p MAX_REPLICAS=${{ env.MAX_REPLICAS_DEV }} -p MIN_CPU=${{ env.MIN_CPU }} -p MAX_CPU=${{ env.MAX_CPU }} -p MIN_MEM=${{ env.MIN_MEM }} -p MAX_MEM=${{ env.MAX_MEM }} \
         | oc apply -f -
         
-        curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/main/tools/config/update-configmap.sh | bash /dev/stdin dev ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }}
-        curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/main/tools/config/update-configmap.sh | bash /dev/stdin dev ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }}
+        curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/main/tools/config/update-configmap.sh | bash /dev/stdin dev ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }} ${{ env.INDEPENDENT_SCHOOLS_NAMESPACE_DEV }}
         
         # Start rollout (if necessary) and follow it
         oc rollout latest dc/${{ env.SPRING_BOOT_IMAGE_NAME }} 2> /dev/null \

.github/workflows/deploy-to.openshift-prod.yml

@@ -9,6 +9,8 @@ env:
   OPENSHIFT_TOKEN: ${{ secrets.OPENSHIFT_TOKEN }}
   OPENSHIFT_NAMESPACE: ${{ secrets.PEN_NAMESPACE_NO_ENV }}-prod
 
+  INDEPENDENT_SCHOOLS_NAMESPACE_PROD: ${{ secrets.INDEPENDENT_SCHOOLS_NAMESPACE_NO_ENV }}-prod
+
   DB_JDBC_CONNECT_STRING: ${{ secrets.DB_JDBC_CONNECT_STRING }}
   DB_PWD: ${{ secrets.DB_PWD }}
   DB_USER: ${{ secrets.DB_USER }}
@@ -121,11 +123,11 @@ jobs:
           oc process -f tools/openshift/api.dc.yaml -p APP_NAME=${{ env.APP_NAME }} -p REPO_NAME=${{ env.REPO_NAME }} -p BRANCH=${{ env.BRANCH }} -p NAMESPACE=${{ env.OPENSHIFT_NAMESPACE }} -p TAG=${{ steps.get-latest-tag.outputs.tag }} -p MIN_REPLICAS=${{ env.MIN_REPLICAS }} -p MAX_REPLICAS=${{ env.MAX_REPLICAS }} -p MIN_CPU=${{ env.MIN_CPU }} -p MAX_CPU=${{ env.MAX_CPU }} -p MIN_MEM=${{ env.MIN_MEM }} -p MAX_MEM=${{ env.MAX_MEM }} \
           | oc apply -f -
           
-          curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/${{ steps.get-latest-tag.outputs.tag }}/tools/config/update-configmap.sh | bash /dev/stdin ${{ env.TARGET_ENV }} ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }}
+          curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/${{ steps.get-latest-tag.outputs.tag }}/tools/config/update-configmap.sh | bash /dev/stdin ${{ env.TARGET_ENV }} ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }} ${{ env.INDEPENDENT_SCHOOLS_NAMESPACE_PROD }}
           
           # Start rollout (if necessary) and follow it
           oc rollout latest dc/${{ env.SPRING_BOOT_IMAGE_NAME }} 2> /dev/null \
           || true && echo "Rollout in progress"
           oc logs -f dc/${{ env.SPRING_BOOT_IMAGE_NAME }}
           # Get status, returns 0 if rollout is successful
-          oc rollout status dc/${{ env.SPRING_BOOT_IMAGE_NAME }}
+          oc rollout status dc/${{ env.SPRING_BOOT_IMAGE_NAME }}

.github/workflows/deploy-to.openshift-test.yml

@@ -9,6 +9,8 @@ env:
   OPENSHIFT_TOKEN: ${{ secrets.OPENSHIFT_TOKEN }}
   OPENSHIFT_NAMESPACE: ${{ secrets.PEN_NAMESPACE_NO_ENV }}-test
 
+  INDEPENDENT_SCHOOLS_NAMESPACE_TEST: ${{ secrets.INDEPENDENT_SCHOOLS_NAMESPACE_NO_ENV }}-test
+
   DB_JDBC_CONNECT_STRING: ${{ secrets.DB_JDBC_CONNECT_STRING }}
   DB_PWD: ${{ secrets.DB_PWD }}
   DB_USER: ${{ secrets.DB_USER }}
@@ -121,7 +123,7 @@ jobs:
           oc process -f tools/openshift/api.dc.yaml -p APP_NAME=${{ env.APP_NAME }} -p REPO_NAME=${{ env.REPO_NAME }} -p BRANCH=${{ env.BRANCH }} -p NAMESPACE=${{ env.OPENSHIFT_NAMESPACE }} -p TAG=${{ steps.get-latest-tag.outputs.tag }} -p MIN_REPLICAS=${{ env.MIN_REPLICAS }} -p MAX_REPLICAS=${{ env.MAX_REPLICAS }} -p MIN_CPU=${{ env.MIN_CPU }} -p MAX_CPU=${{ env.MAX_CPU }} -p MIN_MEM=${{ env.MIN_MEM }} -p MAX_MEM=${{ env.MAX_MEM }} \
           | oc apply -f -
           
-          curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/${{ steps.get-latest-tag.outputs.tag }}/tools/config/update-configmap.sh | bash /dev/stdin ${{ env.TARGET_ENV }} ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }}
+          curl -s https://raw.githubusercontent.com/bcgov/${{ env.REPO_NAME }}/${{ steps.get-latest-tag.outputs.tag }}/tools/config/update-configmap.sh | bash /dev/stdin ${{ env.TARGET_ENV }} ${{ env.APP_NAME }} ${{ env.NAMESPACE }} ${{ env.COMMON_NAMESPACE }} ${{ env.DB_JDBC_CONNECT_STRING }} ${{ env.DB_PWD }} ${{ env.DB_USER }} ${{ env.SPLUNK_TOKEN }} ${{ env.INDEPENDENT_SCHOOLS_NAMESPACE_TEST }}
           
           # Start rollout (if necessary) and follow it
           oc rollout latest dc/${{ env.SPRING_BOOT_IMAGE_NAME }} 2> /dev/null \

api/src/main/java/ca/bc/gov/educ/api/pen/replication/constants/EventOutcome.java

@@ -55,6 +55,12 @@ public enum EventOutcome {
 
   AUTHORITY_CREATED,
 
+  AUTHORITY_CREATED_IN_SPM,
+
+  AUTHORITY_CREATED_IN_IOSAS,
+
+  AUTHORITY_CREATED_IN_ISFS,
+
   AUTHORITY_FOUND,
 
   AUTHORITY_NOT_FOUND

api/src/main/java/ca/bc/gov/educ/api/pen/replication/constants/EventType.java

@@ -50,6 +50,12 @@ public enum EventType {
 
   CREATE_AUTHORITY,
 
+  CREATE_AUTHORITY_IN_SPM,
+
+  CREATE_AUTHORITY_IN_IOSAS,
+
+  CREATE_AUTHORITY_IN_ISFS,
+
   GET_AUTHORITY,
 
   GET_PAGINATED_SCHOOLS,

api/src/main/java/ca/bc/gov/educ/api/pen/replication/constants/IndependentSchoolSystem.java

@@ -0,0 +1,17 @@
+package ca.bc.gov.educ.api.pen.replication.constants;
+
+import lombok.Getter;
+
+
+public enum IndependentSchoolSystem {
+
+  IOSAS("IOSAS"),
+  ISFS("ISFS");
+  @Getter
+  private final String code;
+
+  IndependentSchoolSystem(final String code) {
+    this.code = code;
+  }
+
+}

api/src/main/java/ca/bc/gov/educ/api/pen/replication/constants/SagaEnum.java

@@ -7,21 +7,25 @@
  */
 public enum SagaEnum {
   /**
-   * Pen replication student update saga saga enum.
+   * Pen replication student update saga enum.
    */
   PEN_REPLICATION_STUDENT_UPDATE_SAGA("PEN_REPLICATION_STUDENT_UPDATE_SAGA"),
   /**
-   * Pen replication student create saga saga enum.
+   * Pen replication student create saga enum.
    */
   PEN_REPLICATION_STUDENT_CREATE_SAGA("PEN_REPLICATION_STUDENT_CREATE_SAGA"),
   /**
-   * Pen replication possible match create saga saga enum.
+   * Pen replication possible match create saga enum.
    */
   PEN_REPLICATION_POSSIBLE_MATCH_CREATE_SAGA("PEN_REPLICATION_POSSIBLE_MATCH_CREATE_SAGA"),
   /**
-   * Pen replication possible match delete saga saga enum.
+   * Pen replication possible match delete saga enum.
    */
-  PEN_REPLICATION_POSSIBLE_MATCH_DELETE_SAGA("PEN_REPLICATION_POSSIBLE_MATCH_DELETE_SAGA");
+  PEN_REPLICATION_POSSIBLE_MATCH_DELETE_SAGA("PEN_REPLICATION_POSSIBLE_MATCH_DELETE_SAGA"),
+  /**
+   * Pen replication authority create saga enum.
+   */
+  PEN_REPLICATION_AUTHORITY_CREATE_SAGA("PEN_REPLICATION_AUTHORITY_CREATE_SAGA");
   @Getter
   private final String code;
 

api/src/main/java/ca/bc/gov/educ/api/pen/replication/constants/SagaTopicsEnum.java

@@ -14,6 +14,10 @@ public enum SagaTopicsEnum {
    * Pen replication student create saga topic saga topics enum.
    */
   PEN_REPLICATION_STUDENT_CREATE_SAGA_TOPIC("PEN_REPLICATION_STUDENT_CREATE_SAGA_TOPIC"),
+  /**
+   * Pen replication student create saga topic saga topics enum.
+   */
+  PEN_REPLICATION_AUTHORITY_CREATE_SAGA_TOPIC("PEN_REPLICATION_AUTHORITY_CREATE_SAGA_TOPIC"),
   /**
    * Pen services api topic saga topics enum.
    */

api/src/main/java/ca/bc/gov/educ/api/pen/replication/exception/BusinessError.java

@@ -9,7 +9,8 @@ public enum BusinessError {
   /**
    * The Event already persisted.
    */
-  EVENT_ALREADY_PERSISTED("Event with event id :: $? , is already persisted in DB, a duplicate message from Jet Stream.");
+  EVENT_ALREADY_PERSISTED("Event with event id :: $? , is already persisted in DB, a duplicate message from Jet Stream."),
+  SAGA_ALREADY_PERSISTED("Saga with event id :: $? , is already persisted in DB, a duplicate message from Jet Stream.");
 
   @Getter
   private final String code;

api/src/main/java/ca/bc/gov/educ/api/pen/replication/model/Saga.java

@@ -38,6 +38,11 @@ public class Saga implements Serializable {
   @Column(name = "SAGA_ID", unique = true, updatable = false, columnDefinition = "BINARY(16)")
   UUID sagaId;
 
+  /**
+   * Created by event ID.
+   */
+  @Column(name = "CREATED_FROM_EVENT_ID")
+  UUID createdFromEventID;
 
   /**
    * The Saga name.

api/src/main/java/ca/bc/gov/educ/api/pen/replication/orchestrator/AuthorityCreateOrchestrator.java

@@ -0,0 +1,109 @@
+package ca.bc.gov.educ.api.pen.replication.orchestrator;
+
+import ca.bc.gov.educ.api.pen.replication.constants.IndependentSchoolSystem;
+import ca.bc.gov.educ.api.pen.replication.constants.SagaEnum;
+import ca.bc.gov.educ.api.pen.replication.constants.SagaTopicsEnum;
+import ca.bc.gov.educ.api.pen.replication.mappers.AuthorityMapperHelper;
+import ca.bc.gov.educ.api.pen.replication.mappers.LocalDateTimeMapper;
+import ca.bc.gov.educ.api.pen.replication.messaging.MessagePublisher;
+import ca.bc.gov.educ.api.pen.replication.model.AuthorityMasterEntity;
+import ca.bc.gov.educ.api.pen.replication.model.Saga;
+import ca.bc.gov.educ.api.pen.replication.model.SagaEvent;
+import ca.bc.gov.educ.api.pen.replication.orchestrator.base.BaseOrchestrator;
+import ca.bc.gov.educ.api.pen.replication.repository.AuthorityMasterRepository;
+import ca.bc.gov.educ.api.pen.replication.rest.RestUtils;
+import ca.bc.gov.educ.api.pen.replication.service.AuthorityCreateService;
+import ca.bc.gov.educ.api.pen.replication.service.SagaService;
+import ca.bc.gov.educ.api.pen.replication.struct.Event;
+import ca.bc.gov.educ.api.pen.replication.struct.saga.AuthorityCreateSagaData;
+import ca.bc.gov.educ.api.pen.replication.util.JsonUtil;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import jakarta.persistence.EntityManagerFactory;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.springframework.stereotype.Component;
+
+import static ca.bc.gov.educ.api.pen.replication.constants.EventOutcome.*;
+import static ca.bc.gov.educ.api.pen.replication.constants.EventType.*;
+
+@Component
+@Slf4j
+public class AuthorityCreateOrchestrator extends BaseOrchestrator<AuthorityCreateSagaData> {
+
+  public static final String RESPONDED_VIA_NATS_TO_FOR_EVENT = "responded via NATS to {} for {} Event. :: {}";
+  private final RestUtils restUtils;
+  private final AuthorityCreateService authorityCreateService;
+
+  protected AuthorityCreateOrchestrator(final SagaService sagaService, final MessagePublisher messagePublisher, final EntityManagerFactory entityManagerFactory, final RestUtils restUtils, AuthorityCreateService authorityCreateService) {
+    super(entityManagerFactory, sagaService, messagePublisher, AuthorityCreateSagaData.class, SagaEnum.PEN_REPLICATION_AUTHORITY_CREATE_SAGA, SagaTopicsEnum.PEN_REPLICATION_AUTHORITY_CREATE_SAGA_TOPIC);
+    this.restUtils = restUtils;
+    this.authorityCreateService = authorityCreateService;
+  }
+
+  @Override
+  public void populateStepsToExecuteMap() {
+    this.stepBuilder()
+      .begin(CREATE_AUTHORITY_IN_SPM, this::createAuthorityInSPM)
+      .step(CREATE_AUTHORITY_IN_SPM, AUTHORITY_CREATED_IN_SPM, CREATE_AUTHORITY_IN_IOSAS, this::createAuthorityInIOSAS)
+      .step(CREATE_AUTHORITY_IN_IOSAS, AUTHORITY_CREATED_IN_IOSAS, CREATE_AUTHORITY_IN_ISFS, this::createAuthorityInISFS)
+      .end(CREATE_AUTHORITY_IN_ISFS, AUTHORITY_CREATED_IN_ISFS);
+  }
+
+  private void createAuthorityInSPM(final Event event, final Saga saga, final AuthorityCreateSagaData authorityCreateSagaData) throws JsonProcessingException {
+    saga.setSagaState(CREATE_AUTHORITY_IN_SPM.toString());
+    final SagaEvent eventStates = this.createEventState(saga, event.getEventType(), event.getEventOutcome(), event.getEventPayload());
+    this.getSagaService().updateAttachedSagaWithEvents(saga, eventStates);
+    var authority = authorityCreateSagaData.getIndependentAuthority();
+
+    AuthorityMasterEntity newAuthorityMaster = authorityCreateService.saveAuthority(authority);
+    Event nextEvent = null;
+    if(newAuthorityMaster != null) {
+      nextEvent = Event.builder().sagaId(saga.getSagaId())
+              .eventType(CREATE_AUTHORITY_IN_SPM)
+              .eventOutcome(AUTHORITY_CREATED_IN_SPM)
+              .eventPayload(JsonUtil.getJsonStringFromObject(newAuthorityMaster))
+              .build();
+    }else{
+      nextEvent = Event.builder().sagaId(saga.getSagaId())
+              .eventType(CREATE_AUTHORITY_IN_SPM)
+              .eventOutcome(AUTHORITY_CREATED_IN_SPM)
+              .eventPayload("No update required")
+              .build();
+    }
+    this.postMessageToTopic(this.getTopicToSubscribe().getCode(), nextEvent); // this will make it async and use pub/sub flow even though it is sending message to itself
+    log.info(RESPONDED_VIA_NATS_TO_FOR_EVENT, this.getTopicToSubscribe(), CREATE_AUTHORITY_IN_SPM, saga.getSagaId());
+  }
+
+  private void createAuthorityInIOSAS(final Event event, final Saga saga, final AuthorityCreateSagaData authorityCreateSagaData) throws JsonProcessingException {
+    saga.setSagaState(CREATE_AUTHORITY_IN_IOSAS.toString());
+    final SagaEvent eventStates = this.createEventState(saga, event.getEventType(), event.getEventOutcome(), event.getEventPayload());
+    this.getSagaService().updateAttachedSagaWithEvents(saga, eventStates);
+
+    restUtils.createOrUpdateAuthorityInIndependentSchoolSystem(authorityCreateSagaData.getIndependentAuthority(), IndependentSchoolSystem.IOSAS);
+
+    val nextEvent = Event.builder().sagaId(saga.getSagaId())
+            .eventType(CREATE_AUTHORITY_IN_IOSAS)
+            .eventOutcome(AUTHORITY_CREATED_IN_IOSAS)
+            .eventPayload(JsonUtil.getJsonStringFromObject(authorityCreateSagaData.getIndependentAuthority()))
+            .build();
+    this.postMessageToTopic(this.getTopicToSubscribe().getCode(), nextEvent); // this will make it async and use pub/sub flow even though it is sending message to itself
+    log.info(RESPONDED_VIA_NATS_TO_FOR_EVENT, this.getTopicToSubscribe(), CREATE_AUTHORITY_IN_IOSAS, saga.getSagaId());
+  }
+
+  private void createAuthorityInISFS(final Event event, final Saga saga, final AuthorityCreateSagaData authorityCreateSagaData) throws JsonProcessingException {
+    saga.setSagaState(CREATE_AUTHORITY_IN_ISFS.toString());
+    final SagaEvent eventStates = this.createEventState(saga, event.getEventType(), event.getEventOutcome(), event.getEventPayload());
+    this.getSagaService().updateAttachedSagaWithEvents(saga, eventStates);
+
+    restUtils.createOrUpdateAuthorityInIndependentSchoolSystem(authorityCreateSagaData.getIndependentAuthority(), IndependentSchoolSystem.ISFS);
+
+    val nextEvent = Event.builder().sagaId(saga.getSagaId())
+            .eventType(CREATE_AUTHORITY_IN_ISFS)
+            .eventOutcome(AUTHORITY_CREATED_IN_ISFS)
+            .eventPayload(JsonUtil.getJsonStringFromObject(authorityCreateSagaData.getIndependentAuthority()))
+            .build();
+    this.postMessageToTopic(this.getTopicToSubscribe().getCode(), nextEvent); // this will make it async and use pub/sub flow even though it is sending message to itself
+    log.info(RESPONDED_VIA_NATS_TO_FOR_EVENT, this.getTopicToSubscribe(), CREATE_AUTHORITY_IN_ISFS, saga.getSagaId());
+  }
+
+}

api/src/main/java/ca/bc/gov/educ/api/pen/replication/properties/ApplicationProperties.java

@@ -65,4 +65,7 @@ public class ApplicationProperties {
 
   @Value("${url.api.institute}")
   private String instituteApiURL;
+
+  @Value("${url.api.independent.schools}")
+  private String independentSchoolsAPI;
 }

api/src/main/java/ca/bc/gov/educ/api/pen/replication/repository/SagaRepository.java

@@ -10,29 +10,18 @@
 
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 
 /**
  * The interface Saga repository.
  */
 @Repository
 public interface SagaRepository extends JpaRepository<Saga, UUID>, JpaSpecificationExecutor<Saga> {
-  /**
-   * Find all by status in list.
-   *
-   * @param statuses the statuses
-   * @return the list
-   */
-  List<Saga> findAllByStatusInOrderByCreateDate(List<String> statuses);
 
+  List<Saga> findAllByStatusInOrderByCreateDate(List<String> statuses);
 
-  /**
-   * Find all by create date before list.
-   *
-   * @param createDate the create date
-   * @return the list
-   */
-  List<Saga> findAllByCreateDateBefore(LocalDateTime createDate);
+  Optional<Saga> findByCreatedFromEventID(UUID createdFromEventID);
 
   long countSagasBySagaNameInAndStatusIn(List<String> sagaNames, List<String> statuses);
 

api/src/main/java/ca/bc/gov/educ/api/pen/replication/rest/RestUtils.java

@@ -2,6 +2,7 @@
 
 import ca.bc.gov.educ.api.pen.replication.constants.EventOutcome;
 import ca.bc.gov.educ.api.pen.replication.constants.EventType;
+import ca.bc.gov.educ.api.pen.replication.constants.IndependentSchoolSystem;
 import ca.bc.gov.educ.api.pen.replication.exception.PenReplicationAPIRuntimeException;
 import ca.bc.gov.educ.api.pen.replication.filter.FilterOperation;
 import ca.bc.gov.educ.api.pen.replication.messaging.MessagePublisher;
@@ -25,6 +26,7 @@
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
 
 import java.io.IOException;
 import java.net.URLEncoder;
@@ -236,6 +238,16 @@ public List<IndependentAuthority> getAuthoritiesForOpeningAndClosing(UUID correl
     }
   }
 
+  public void createOrUpdateAuthorityInIndependentSchoolSystem(final IndependentAuthority authority, final IndependentSchoolSystem system) {
+     this.webClient.post()
+            .uri(this.props.getIndependentSchoolsAPI() + "/api/School/" + system.getCode() + "/AuthorityUpsert")
+            .header(CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
+            .body(Mono.just(authority), IndependentAuthority.class)
+            .retrieve()
+            .bodyToMono(String.class)
+            .block();
+  }
+
   private SearchCriteria getCriteria(final String key, final FilterOperation operation, final String value, final ValueType valueType) {
     return SearchCriteria.builder().key(key).operation(operation).value(value).valueType(valueType).build();
   }

api/src/main/java/ca/bc/gov/educ/api/pen/replication/service/AuthorityCreateService.java

@@ -2,6 +2,7 @@
 
 import ca.bc.gov.educ.api.pen.replication.mappers.AuthorityMapperHelper;
 import ca.bc.gov.educ.api.pen.replication.mappers.LocalDateTimeMapper;
+import ca.bc.gov.educ.api.pen.replication.model.AuthorityMasterEntity;
 import ca.bc.gov.educ.api.pen.replication.model.Event;
 import ca.bc.gov.educ.api.pen.replication.repository.AuthorityMasterRepository;
 import ca.bc.gov.educ.api.pen.replication.repository.EventRepository;
@@ -12,6 +13,8 @@
 import lombok.val;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 
@@ -46,6 +49,18 @@ public void processEvent(final IndependentAuthority authority, final Event event
     this.updateEvent(event);
   }
 
+  @Transactional(propagation = Propagation.REQUIRES_NEW)
+  public AuthorityMasterEntity saveAuthority(final IndependentAuthority authority) {
+    if (StringUtils.isNotEmpty(authority.getOpenedDate()) && dateTimeMapper.map(authority.getOpenedDate()).isBefore(LocalDateTime.now())){
+      val existingSchoolMasterRecord = this.authorityMasterRepository.findById(authority.getAuthorityNumber());
+      if (!existingSchoolMasterRecord.isPresent()) {
+        var newAuthorityMaster = authorityMapperHelper.toAuthorityMaster(authority, true);
+        return authorityMasterRepository.save(newAuthorityMaster);
+      }
+    }
+    return null;
+  }
+
   @Override
   public String getEventType() {
     return CREATE_AUTHORITY.toString();