[backend] feat(wf-engine): init split service layer (#4842)

Co-authored-by: EvaE-Filigran <eva.ezzeribi@filigran.io>
Co-authored-by: Camille Roux <camille.roux@filigran.io>
Co-authored-by: savacano28 <stephanya.casanova@filigran.io>

Author: 4 people

openaev-api/src/main/java/io/openaev/aop/WorkflowUpdateEventAspect.java

@@ -0,0 +1,177 @@
+package io.openaev.aop;
+
+import io.openaev.rest.settings.PreviewFeature;
+import io.openaev.service.PreviewFeatureService;
+import io.openaev.service.chaining.QueueChainingService;
+import io.openaev.service.chaining.StepService;
+import java.io.IOException;
+import java.util.*;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.aspectj.lang.JoinPoint;
+import org.aspectj.lang.annotation.AfterReturning;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.reflect.MethodSignature;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.stereotype.Component;
+
+/**
+ * Aspect that intercepts methods annotated with {@link WorkflowUpdateEvent} to trigger workflow
+ * updates in the chaining engine.
+ *
+ * <p>This aspect uses SpEL expressions defined in the annotation to extract inject or expectation
+ * IDs from method parameters, then sends update events to the workflow external update queue.
+ */
+@Aspect
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class WorkflowUpdateEventAspect {
+
+  /**
+   * This list will contain IDs of any event that couldn't be sent due to a problem with the queue
+   * (most likely a network error, or the queue system being down)
+   */
+  private Set<String> unsentEventsCache = new HashSet<>();
+
+  private final PreviewFeatureService previewFeatureService;
+
+  private final QueueChainingService queueChainingService;
+  private final StepService stepService;
+
+  private final ExpressionParser parser = new SpelExpressionParser();
+
+  /**
+   * Advice executed after methods annotated with {@link WorkflowUpdateEvent} returns. If an
+   * exception is throw by the annotated method, this advice will not be called.
+   *
+   * <p>Extracts the inject ID or expectation IDs from method parameters using SpEL expressions
+   * defined in the annotation, then dispatches workflow update events accordingly.
+   *
+   * @param joinPoint the join point providing access to the method signature and arguments
+   * @param annotation the {@link WorkflowUpdateEvent} annotation containing SpEL expressions
+   * @throws IllegalStateException if the annotation does not specify exactly one of injectId or
+   *     expectationIds
+   */
+  @AfterReturning("@annotation(annotation)")
+  public void afterEventProcessed(JoinPoint joinPoint, WorkflowUpdateEvent annotation) {
+    if (!previewFeatureService.isFeatureEnabled(PreviewFeature.INJECT_CHAINING)) {
+      // No chaining enabled, does nothing
+      return;
+    }
+
+    String injectIdSPEL = annotation.injectId();
+    String expectationIdsSPEL = annotation.expectationIds();
+
+    boolean hasInjectId = StringUtils.isNotBlank(injectIdSPEL);
+    boolean hasExpectation = StringUtils.isNotBlank(expectationIdsSPEL);
+
+    if (hasInjectId == hasExpectation) {
+      throw new IllegalStateException(
+          "Annotation @WorkflowUpdateEvent on "
+              + joinPoint.getSignature().toShortString()
+              + " must set exactly one of injectId or expectationIds");
+    }
+
+    MethodSignature signature = (MethodSignature) joinPoint.getSignature();
+    String[] parameterNames = signature.getParameterNames();
+    Object[] args = joinPoint.getArgs();
+
+    // Create SpEL evaluation context to retrieve the resource ID if it exists
+    EvaluationContext context = new StandardEvaluationContext();
+
+    // Add all method parameters to context
+    for (int i = 0; i < parameterNames.length; i++) {
+      context.setVariable(parameterNames[i], args[i]);
+    }
+
+    if (hasInjectId) {
+      this.handleInjectIdParam(context, injectIdSPEL);
+    } else {
+      this.handleExpectationTracesParam(context, expectationIdsSPEL);
+    }
+
+    // Retry events that are stored in the unsent cache due to previous errors
+    sendEvents(unsentEventsCache);
+  }
+
+  /**
+   * Send a workflow update event related to the given inject to the queue, and update the unsent
+   * events cache if an error occurs
+   *
+   * @param context the SpEL evaluation context
+   * @param injectIdSPEL the SpEL expression to fetch the injectId from the request
+   */
+  private void handleInjectIdParam(EvaluationContext context, String injectIdSPEL) {
+    Expression exp = parser.parseExpression(injectIdSPEL);
+    String injectId =
+        exp.getValue(context) != null
+            ? Objects.requireNonNull(exp.getValue(context)).toString()
+            : "";
+
+    if (!injectId.isEmpty()) {
+      String stepId = stepService.findStepIdByInjectId(injectId);
+      try {
+        queueChainingService.updateStep(stepId);
+      } catch (IOException e) {
+        // In case an error occurs, we store the inject in the unsent event cache to be retried
+        // later, when other events will be sent
+        unsentEventsCache.add(stepId);
+      }
+    }
+  }
+
+  /**
+   * Send a workflow update event related to all the injects related to the given expectation IDs to
+   * the queue
+   *
+   * @param context the SpEL evaluation context
+   * @param expectationIDsdSPEL the SpEL expression to fetch the injectId from the request
+   */
+  private void handleExpectationTracesParam(EvaluationContext context, String expectationIDsdSPEL) {
+    Expression exp = parser.parseExpression(expectationIDsdSPEL);
+    Object expectationIdsFromSPEL =
+        exp.getValue(context) != null ? Objects.requireNonNull(exp.getValue(context)) : null;
+
+    Set<String> expectationIds = new HashSet<>();
+    if (expectationIdsFromSPEL instanceof Collection<?> c) {
+      c.stream().map(Object::toString).forEach(expectationIds::add);
+    } else if (expectationIdsFromSPEL instanceof String expectationId) {
+      expectationIds.add(expectationId);
+    } else {
+      throw new IllegalStateException(
+          "@WorkflowUpdateEvent.expectationIDsdSpEL must return a Collection or a String");
+    }
+
+    Set<String> stepIds = stepService.findStepIdsByExpectationIds(expectationIds);
+    sendEvents(stepIds);
+  }
+
+  /**
+   * Send a list of events in the queue, and update the unsent events cache if an error occurs
+   *
+   * @param stepIds step IDs to notify with an event
+   */
+  private void sendEvents(Set<String> stepIds) {
+    if (stepIds.isEmpty()) {
+      return;
+    }
+    Set<String> remainingUnsetEvents = new HashSet<>(stepIds);
+    try {
+      for (String stepId : stepIds) {
+        queueChainingService.updateStep(stepId);
+        remainingUnsetEvents.remove(stepId);
+      }
+      unsentEventsCache.clear();
+    } catch (IOException e) {
+      // If a fail occurs, no need to continue the loop, just keep the remaining events in the
+      // unsent events cache to be retried later, when other events will be sent
+      unsentEventsCache = remainingUnsetEvents;
+    }
+  }
+}

openaev-api/src/main/java/io/openaev/api/chaining/ActionStep.java

@@ -0,0 +1,57 @@
+package io.openaev.api.chaining;
+
+import io.openaev.api.chaining.dto.StepsCreateInput;
+import io.openaev.database.model.Step;
+import io.openaev.database.model.Workflow;
+import io.openaev.rest.exception.ChainingException;
+import java.util.Optional;
+
+/** The interface Action step. IMPLEMENTED BY: - InjectExecutionStep */
+public interface ActionStep {
+  /**
+   * Create step.
+   *
+   * @param stepInput the step input
+   * @param workflow the workflow
+   * @return the step
+   */
+  Optional<Step> create(StepsCreateInput.StepCreateInput stepInput, Workflow workflow)
+      throws ChainingException;
+
+  /**
+   * Creates a Ready step. The step is created with status READY based on a step template.
+   * Duplicates the template step and fills its content from the input.
+   *
+   * @param stepTemplate the stepTemplate
+   * @param input the input for the new step
+   * @param workflowRun the workflow run
+   * @return the created Ready step
+   */
+  Optional<Step> ready(Step stepTemplate, String input, Workflow workflowRun)
+      throws ChainingException;
+
+  /**
+   * Executes a Ready step. Changes the status from READY to RUN.
+   *
+   * @param readyStep the step currently in READY status
+   * @return the step after being set to RUN
+   */
+  Optional<Step> run(Step readyStep) throws ChainingException;
+
+  /**
+   * Updates a step. Applies the necessary processing based on the new output.
+   *
+   * @param stepRun the step run to update
+   * @return the updated step
+   */
+  Optional<Step> update(Step stepRun) throws ChainingException;
+
+  /**
+   * Ends a step. Checks if all expected outputs have been received and updates the status from RUN
+   * to END.
+   *
+   * @param stepRun the step to end
+   * @param workflow the workflow
+   */
+  void end(Step stepRun, Workflow workflow) throws ChainingException;
+}

openaev-api/src/main/java/io/openaev/api/chaining/DataInputStep.java

@@ -0,0 +1,28 @@
+package io.openaev.api.chaining;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.openaev.rest.inject.form.InjectInput;
+
+/**
+ * Interface representing the data associated with a step.
+ *
+ * <p>This is a polymorphic type used to handle different kinds of step data. The concrete type is
+ * determined by the JSON property {@code "type"}.
+ *
+ * <p>Currently supported implementations:
+ *
+ * <ul>
+ *   <li>{@link InjectInput} with type name "inject"
+ * </ul>
+ *
+ * <p>When serialized/deserialized with Jackson, the {@code type} property determines which concrete
+ * implementation to use.
+ */
+@JsonTypeInfo(
+    use = JsonTypeInfo.Id.NAME,
+    include = JsonTypeInfo.As.PROPERTY,
+    property = "type",
+    defaultImpl = InjectInput.class)
+@JsonSubTypes({@JsonSubTypes.Type(value = InjectInput.class, name = "inject")})
+public interface DataInputStep {}