package eu.reply.lea.services.yms.task.queue;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import javax.persistence.AttributeOverride;
import javax.persistence.AttributeOverrides;
import javax.persistence.CascadeType;
import javax.persistence.Column;
import javax.persistence.Embedded;
import javax.persistence.Entity;
import javax.persistence.Index;
import javax.persistence.Inheritance;
import javax.persistence.InheritanceType;
import javax.persistence.OneToMany;
import javax.persistence.Table;

import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Getter;

import eu.reply.lea.services.masterdata.reference.LocationReference;
import eu.reply.lea.services.yms.task.queue.event.ElementQueueProcessCompletedEvent;
import eu.reply.lea.services.yms.task.queue.event.ElementQueueProcessPartialConfirmEvent;
import eu.reply.lea.services.yms.util.AbstractProcess;
import eu.reply.lea.services.yms.util.ProcessReference;
import eu.reply.sce.backend.core.entity.EntityReference;
import eu.reply.sce.backend.core.jee.event.Producer;

@Entity
@Inheritance(strategy = InheritanceType.SINGLE_TABLE)
@Table(indexes = {
@Index(name="idx_process_external_entity_status", columnList = "externalEntityRef_entity_id, externalEntityRef_entity_name, externalEntityRef_module, processCompletionType")
})
public abstract class ElementQueueProcess extends AbstractProcess {


@OneToMany(mappedBy = "elementQueueProcess", cascade = CascadeType.ALL)
private final List<QueueProcessItem> queueProcessItems = new ArrayList<>();

@Getter
@Embedded
@AttributeOverrides({
    @AttributeOverride(name = "module", column = @Column(name = "externalEntityRef_module")),
    @AttributeOverride(name = "entityName", column = @Column(name = "externalEntityRef_entity_name")),
    @AttributeOverride(name = "entityId", column = @Column(name = "externalEntityRef_entity_id")),
    @AttributeOverride(name = "businessKey", column = @Column(name = "externalEntityRef_business_key")),
    @AttributeOverride(name = "usageType", column = @Column(name = "externalEntityRef_usage_type"))
})
private EntityReference externalEntityReference;

protected ElementQueueProcess() {
    super();
}

public ElementQueueProcess(final EntityReference externalEntityReference){
    this.externalEntityReference = externalEntityReference;
}

public List<QueueProcessItem> getQueueProcessItems() {
    return Collections.unmodifiableList(this.queueProcessItems);
}

/**
 * Confirms the process at a specific location and fires events based on the process state.
 *
 * @param locationReference                              The location to confirm.
 * @param elementQueueProcessPartialConfirmEventProducer Producer for firing partial confirmation events.
 * @param elementQueueProcessCompletedEventProducer      Producer for firing completion events.
 * @throws IllegalStateException if the process has already been completed.
 */
public void confirm(final LocationReference locationReference,
                    final Producer<ElementQueueProcessPartialConfirmEvent> elementQueueProcessPartialConfirmEventProducer,
                    final Producer<ElementQueueProcessCompletedEvent> elementQueueProcessCompletedEventProducer) {
    if (this.isCompleted()) {
        throw new IllegalStateException("The process has already been completed");
    }
    boolean confirmed = this.confirmElement(locationReference);
    this.completeProcess(ProcessCompletionType.CompletedNormally);
    if (!this.isCompleted() && confirmed) {
        elementQueueProcessPartialConfirmEventProducer.fire(
            new ElementQueueProcessPartialConfirmEvent(new ProcessReference(this)));
    }
    if (this.isCompleted()) {
        elementQueueProcessCompletedEventProducer.fire(
            new ElementQueueProcessCompletedEvent(new ProcessReference(this), externalEntityReference,
                                                  this.getProcessCompletionType()));
    }
}

/**
 * Aborts the process at a specific location and fires a completion event if the process is completed.
 *
 * @param executionLocation                         The location where the process should be aborted.
 * @param elementQueueProcessCompletedEventProducer Producer for firing completion events.
 * @throws IllegalStateException if the process has already been completed.
 */
public void abort(final LocationReference executionLocation,
                  final Producer<ElementQueueProcessCompletedEvent> elementQueueProcessCompletedEventProducer) {
    if (this.isCompleted()) {
        throw new IllegalStateException("The process has already been completed");
    }
    this.queueProcessItems.stream().filter(
        queueProcessItem -> queueProcessItem.getExecutionLocation().equals(executionLocation)).forEach(
        queueProcessItem -> queueProcessItem.abort()
    );
    this.abortProcess(ProcessCompletionType.Aborted);
    if (this.isCompleted()) {
        elementQueueProcessCompletedEventProducer.fire(
            new ElementQueueProcessCompletedEvent(new ProcessReference(this), externalEntityReference,
                                                  this.getProcessCompletionType()));
    }
}

private boolean confirmElement(final LocationReference locationReference) {
    Long numberOfInitialConfirmedElements = this.numberOfConfirmations();
    this.queueProcessItems.stream().filter(
        queueProcessItem -> queueProcessItem.getExecutionLocation().equals(locationReference)).forEach(
        queueProcessItem -> queueProcessItem.confirm()
    );
    Long numberOfFinalConfirmedElements = this.numberOfConfirmations();
    return numberOfFinalConfirmedElements.compareTo(numberOfInitialConfirmedElements) > 0;
}

private Long numberOfConfirmations() {
    return this.queueProcessItems.stream().filter(QueueProcessItem::isConfirmed).count();
}

/**
 * Add a new execution location.
 *
 * @param executionLocations List of locations to be added.
 * @param scheduled boolean to indicate if the queueing was scheduled
 * @return A read-only list of location statuses.
 */
public boolean addExecutionLocation(final LocationReference executionLocations,
                                             final boolean scheduled) {
    return queueProcessItems.add(new QueueProcessItem(this,
                                                      executionLocations, scheduled,
                                                      queueProcessItems.size()));
}



/**
 * Verifies if all element locations within the process have been confirmed.
 * It is a crucial part of the process lifecycle, ensuring that no further action is required for this process
 * to be considered complete.
 *
 * @return true if all element location statuses are confirmed, false otherwise.
 */
@Override
protected boolean completenessVerified() {
    return queueProcessItems.stream().allMatch(QueueProcessItem::isConfirmed);
}

@Override
protected boolean abortionVerified() {
    return queueProcessItems.stream().allMatch(el -> el.isAborted() || !el.isStarted());
}

/**
 * Retrieves the next Item containing information about the next queue location.
 *
 * @return The next QueueProcessItem to be confirmed, or null if all locations are confirmed.
 */
@JsonIgnore
public QueueProcessItem getNextQueueProcessItem() {
    QueueProcessItem queueProcessItem = null;

    for (QueueProcessItem status : this.queueProcessItems) {
        if (!status.isConfirmed() && (queueProcessItem == null
                                      || status.getEnqueueOrder() < queueProcessItem.getEnqueueOrder())) {
            queueProcessItem = status;
        }
    }

    return queueProcessItem;
}

/**
 * Marks the queueing process as started for a specific location.
 *
 * @param executionLocation The location where queueing has started.
 */
public void started(final LocationReference executionLocation) {
    for (QueueProcessItem queueProcessItem : this.queueProcessItems) {
        if (queueProcessItem.getExecutionLocation().equals(executionLocation)) {
            queueProcessItem.start();
        }
    }
}
}
package eu.reply.lea.services.yms.util;

import javax.persistence.MappedSuperclass;
import java.time.Duration;
import java.time.Instant;

import eu.reply.sce.backend.core.entity.StandardEntity;

@MappedSuperclass
public abstract class AbstractProcess extends StandardEntity implements Process{


private static final long serialVersionUID = 1L;
private ProcessCompletionType processCompletionType;
private Instant startTime;

public AbstractProcess() {

    super();

    this.setProcessCompletionType(ProcessCompletionType.NotCompleted);
    this.setStartTime(Instant.now());
}




@Override
public Duration currentDuration() {
    return this.calculateTotalCurrentDuration(Instant.now());
}

@Override
public boolean isCompleted() {
    return !this.notCompleted();
}

@Override
public boolean notCompleted() {
    return this.getProcessCompletionType().equals(ProcessCompletionType.NotCompleted);
}

@Override
public boolean isAborted(){
    return this.getProcessCompletionType().equals(ProcessCompletionType.Aborted);
}

@Override
public boolean didProcessingComplete() {
    return this.isCompleted() && !this.isAborted();
}

@Override
public ProcessCompletionType getProcessCompletionType() {
    return this.processCompletionType;
}


@Override
public Instant getStartTime() {
    return this.startTime;
}



@Override
public String toString() {
    return "AbstractProcess [id=" + this.getId()
           + ", processCompletionType=" + processCompletionType + ", startTime=" + startTime + "]";
}

protected void completeProcess(ProcessCompletionType aProcessCompletionType) {
    if (!this.isCompleted() && this.completenessVerified()) {
        this.setProcessCompletionType(aProcessCompletionType);
    }
}

protected void abortProcess(ProcessCompletionType aProcessCompletionType) {
    if (!this.isCompleted() && this.abortionVerified()) {
        this.setProcessCompletionType(aProcessCompletionType);
    }
}


protected abstract boolean completenessVerified();
protected abstract boolean abortionVerified();

private Duration calculateTotalCurrentDuration(Instant aInstantFollowingStartTime) {
    return Duration.between(this.getStartTime(),aInstantFollowingStartTime);
}

private void setProcessCompletionType(ProcessCompletionType aProcessCompletionType) {
    this.processCompletionType = aProcessCompletionType;
}


private void setStartTime(Instant aStartTime) {
    this.startTime = aStartTime;
}
}
package eu.reply.lea.services.yms.task.queue;

import java.util.List;
import javax.persistence.Entity;
import javax.persistence.NamedQueries;
import javax.persistence.NamedQuery;

import lombok.NoArgsConstructor;

import eu.reply.lea.services.masterdata.reference.LocationReference;
import eu.reply.sce.backend.core.entity.EntityReference;

/**

This class handles the dequeue of the elements based on the location provided.
*/
@Entity
@NoArgsConstructor
@NamedQueries(value = {
@NamedQuery(name = ElementDequeueProcess.Q_FIND_BY_QUEUED_ELEMENT_AND_COMPLETENESS, query =
"select eq from ElementDequeueProcess eq left join QueuedElement el "
+ "on el.externalEntityRef.entityId = eq.externalEntityReference.entityId "
+ "and el.externalEntityRef.entityName = eq.externalEntityReference.entityName "
+ "and el.externalEntityRef.module = eq.externalEntityReference.module "
+ "where el.id = :elementQueueId and eq.processCompletionType = :processCompletionType")
})
public class ElementDequeueProcess extends ElementQueueProcess {


public static final String Q_FIND_BY_QUEUED_ELEMENT_AND_COMPLETENESS = "ElementDequeueProcess.Q_FIND_BY_QUEUED_ELEMENT_AND_COMPLETENESS";

public ElementDequeueProcess(final EntityReference externalEntityReference, final List<LocationReference> executionLocations) {
    super(externalEntityReference);
    for (LocationReference executionLocation : executionLocations){
        this.addExecutionLocation(executionLocation, false);
    }
}
}

package eu.reply.lea.services.yms.task.queue;

import javax.persistence.Entity;
import javax.persistence.NamedQueries;
import javax.persistence.NamedQuery;

import lombok.NoArgsConstructor;

import eu.reply.sce.backend.core.entity.EntityReference;

/**

This class handles the enqueue of the elements based on the location provided.
*/
@Entity
@NoArgsConstructor
@NamedQueries(value = {
@NamedQuery(name = ElementEnqueueProcess.Q_FIND_BY_QUEUED_ELEMENT_AND_COMPLETENESS, query =
"select eq from ElementEnqueueProcess eq left join QueuedElement el "
+ "on el.externalEntityRef.entityId = eq.externalEntityReference.entityId "
+ "and el.externalEntityRef.entityName = eq.externalEntityReference.entityName "
+ "and el.externalEntityRef.module = eq.externalEntityReference.module "
+ "where el.id = :elementQueueId and eq.processCompletionType = :processCompletionType")
})
public class ElementEnqueueProcess extends ElementQueueProcess {


public static final String Q_FIND_BY_QUEUED_ELEMENT_AND_COMPLETENESS = "ElementEnqueueProcess.Q_FIND_BY_QUEUED_ELEMENT_AND_COMPLETENESS";



public ElementEnqueueProcess(final EntityReference externalEntityReference) {
    super(externalEntityReference);
}
}

I have the above, but i have now realized that i may want to create a process that does both dequeue and enqueue, following ddd principles how can i do that?
This can be considered a saga pattern as Vaughn Vernon intended:
The above code is already in production, so you have to consider that any modification will affect the existing code