feat: Implement correlation on event filter

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/AbstractEventFilterBuilder.java

@@ -36,9 +36,13 @@ public SELF with(Consumer<P> c) {
     return self();
   }
 
-  public SELF correlate(String key, Consumer<ListenTaskBuilder.CorrelatePropertyBuilder> c) {
-    throw new UnsupportedOperationException(
-        "correlate is not supported in the engine level: https://github.com/serverlessworkflow/sdk-java/issues/1206");
+  public SELF correlate(
+      String key, Consumer<? super AbstractListenTaskBuilder.CorrelatePropertyBuilder> c) {
+    AbstractListenTaskBuilder.CorrelatePropertyBuilder cb =
+        new AbstractListenTaskBuilder.CorrelatePropertyBuilder();
+    c.accept(cb);
+    correlate.setAdditionalProperty(key, cb.build());
+    return self();
   }
 
   public EventFilter build() {

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/AbstractEventFilterSpec.java

@@ -17,6 +17,7 @@
 
 import io.serverlessworkflow.fluent.spec.AbstractEventFilterBuilder;
 import io.serverlessworkflow.fluent.spec.AbstractEventPropertiesBuilder;
+import io.serverlessworkflow.fluent.spec.AbstractListenTaskBuilder;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
@@ -41,13 +42,11 @@ protected List<Consumer<EVENT_FILTER>> getFilterSteps() {
     return filterSteps;
   }
 
-  // TODO: "correlate is not supported in the engine level:
-  // https://github.com/serverlessworkflow/sdk-java/issues/1206". Keeping the code for a future
-  // reference.
-  //  public SELF correlate(String key, Consumer<ListenTaskBuilder.CorrelatePropertyBuilder> c) {
-  //    filterSteps.add(f -> f.correlate(key, c));
-  //    return self();
-  //  }
+  public SELF correlate(
+      String key, Consumer<? super AbstractListenTaskBuilder.CorrelatePropertyBuilder> c) {
+    addFilterStep(f -> f.correlate(key, c));
+    return self();
+  }
 
   @Override
   public void accept(EVENT_FILTER filterBuilder) {

fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java

@@ -37,6 +37,7 @@
 import io.serverlessworkflow.api.types.AuthenticationPolicyUnion;
 import io.serverlessworkflow.api.types.CallHTTP;
 import io.serverlessworkflow.api.types.CatchErrors;
+import io.serverlessworkflow.api.types.CorrelateProperty;
 import io.serverlessworkflow.api.types.Document;
 import io.serverlessworkflow.api.types.EmitEventDefinition;
 import io.serverlessworkflow.api.types.EmitTask;
@@ -310,8 +311,12 @@ void testDoTaskListenOne() {
                                 to ->
                                     to.one(
                                         f ->
-                                            f.with(
-                                                p -> p.type("com.fake.pet").source("mySource"))))))
+                                            f.with(p -> p.type("com.fake.pet").source("mySource"))
+                                                .correlate(
+                                                    "orderId",
+                                                    c ->
+                                                        c.from("$.data.orderId")
+                                                            .expect("$.input.orderId"))))))
             .build();
 
     List<TaskItem> items = wf.getDo();
@@ -327,6 +332,10 @@ void testDoTaskListenOne() {
     EventFilter filter = one.getOne();
     assertNotNull(filter, "EventFilter should be present");
     assertEquals("com.fake.pet", filter.getWith().getType(), "Filter type should match");
+    CorrelateProperty correlate = filter.getCorrelate().getAdditionalProperties().get("orderId");
+    assertNotNull(correlate, "Correlate property should be present");
+    assertEquals("$.data.orderId", correlate.getFrom(), "Correlate from should match");
+    assertEquals("$.input.orderId", correlate.getExpect(), "Correlate expect should match");
   }
 
   @Test

fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/DSLTest.java

@@ -30,6 +30,7 @@
 import static io.serverlessworkflow.fluent.spec.dsl.DSL.workflow;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import io.serverlessworkflow.api.types.CorrelateProperty;
 import io.serverlessworkflow.api.types.HTTPArguments;
 import io.serverlessworkflow.api.types.ListenTaskConfiguration;
 import io.serverlessworkflow.api.types.RunTaskConfiguration;
@@ -166,7 +167,15 @@ public void when_listen_any_with_until() {
   public void when_listen_one() {
     Workflow wf =
         WorkflowBuilder.workflow("f", "ns", "1")
-            .tasks(t -> t.listen(to().one(event().type("only-once"))))
+            .tasks(
+                t ->
+                    t.listen(
+                        to().one(
+                                event()
+                                    .type("only-once")
+                                    .correlate(
+                                        "workflowInstanceId",
+                                        c -> c.from("$.metadata.instanceId")))))
             .build();
 
     var to = wf.getDo().get(0).getTask().getListenTask().getListen().getTo();
@@ -178,6 +187,10 @@ public void when_listen_one() {
     var one = to.getOneEventConsumptionStrategy().getOne();
     assertThat(one.getWith()).isNotNull();
     assertThat(one.getWith().getType()).isEqualTo("only-once");
+    CorrelateProperty correlate =
+        one.getCorrelate().getAdditionalProperties().get("workflowInstanceId");
+    assertThat(correlate).isNotNull();
+    assertThat(correlate.getFrom()).isEqualTo("$.metadata.instanceId");
   }
 
   @Test

impl/core/src/main/java/io/serverlessworkflow/impl/CompositeExpressionFactory.java

@@ -49,6 +49,11 @@ public WorkflowValueResolver<String> resolveString(ExpressionDescriptor desc) {
     return processFactories(desc, f -> f.resolveString(desc));
   }
 
+  @Override
+  public WorkflowValueResolver<Object> resolveValue(ExpressionDescriptor desc) {
+    return processFactories(desc, f -> f.resolveValue(desc));
+  }
+
   @Override
   public WorkflowValueResolver<OffsetDateTime> resolveDate(ExpressionDescriptor desc) {
     return processFactories(desc, f -> f.resolveDate(desc));

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java

@@ -52,4 +52,6 @@ public interface WorkflowInstance extends WorkflowInstanceData {
   boolean resume();
 
   <T> T addMetadataIfAbsent(String key, Supplier<T> supplier);
+
+  void removeMetadataIfPresent(String key);
 }

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java

@@ -355,6 +355,11 @@ public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
   }
 
+  @Override
+  public void removeMetadataIfPresent(String key) {
+    additionalObjects.remove(key);
+  }
+
   @Override
   public <T> Optional<T> findMetadata(String key, Class<T> objectClass) {
     Object value = additionalObjects.get(key);

impl/core/src/main/java/io/serverlessworkflow/impl/events/AbstractTypeConsumer.java

@@ -16,12 +16,16 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.api.types.CorrelateProperty;
 import io.serverlessworkflow.api.types.EventFilter;
+import io.serverlessworkflow.api.types.EventFilterCorrelate;
 import io.serverlessworkflow.api.types.EventProperties;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowModel;
 import java.util.AbstractCollection;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
@@ -37,6 +41,8 @@ public abstract class AbstractTypeConsumer
 
   private static final Logger logger = LoggerFactory.getLogger(AbstractTypeConsumer.class);
 
+  private static final CloudEventPredicate ALWAYS_TRUE = (ce, wf, t) -> true;
+
   protected abstract void registerToAll(Consumer<CloudEvent> consumer);
 
   protected abstract void unregisterFromAll();
@@ -52,13 +58,32 @@ public TypeEventRegistrationBuilder listen(
       EventFilter register, WorkflowApplication application) {
     EventProperties properties = register.getWith();
     String type = properties.getType();
-    return new TypeEventRegistrationBuilder(
-        type, application.cloudEventPredicateFactory().build(application, properties));
+    CloudEventPredicate cePredicate =
+        application.cloudEventPredicateFactory().build(application, properties);
+    Collection<CloudEventPredicate> correlationPredicates =
+        buildCorrelationPredicates(register.getCorrelate(), application);
+    return new TypeEventRegistrationBuilder(type, cePredicate, correlationPredicates);
+  }
+
+  private Collection<CloudEventPredicate> buildCorrelationPredicates(
+      EventFilterCorrelate correlate, WorkflowApplication application) {
+    if (correlate == null) {
+      return List.of();
+    }
+    Map<String, CorrelateProperty> additionalProperties = correlate.getAdditionalProperties();
+    if (additionalProperties == null || additionalProperties.isEmpty()) {
+      return List.of();
+    }
+    Collection<CloudEventPredicate> predicates = new ArrayList<>();
+    for (Map.Entry<String, CorrelateProperty> entry : additionalProperties.entrySet()) {
+      predicates.add(CorrelationPredicate.from(entry.getKey(), entry.getValue(), application));
+    }
+    return predicates;
   }
 
   @Override
   public Collection<TypeEventRegistrationBuilder> listenToAll(WorkflowApplication application) {
-    return List.of(new TypeEventRegistrationBuilder(null, null));
+    return List.of(new TypeEventRegistrationBuilder(null, ALWAYS_TRUE, List.of()));
   }
 
   private static class CloudEventConsumer extends AbstractCollection<TypeEventRegistration>
@@ -68,11 +93,42 @@ private static class CloudEventConsumer extends AbstractCollection<TypeEventRegi
     @Override
     public void accept(CloudEvent ce) {
       logger.debug("Received cloud event {}", ce);
+      WorkflowModel eventModel = null;
       for (TypeEventRegistration registration : registrations) {
-        if (registration.predicate().test(ce, registration.workflow(), registration.task())) {
-          registration.consumer().accept(ce);
+        if (!registration.predicate().test(ce, registration.workflow(), registration.task())) {
+          continue;
+        }
+        Collection<CloudEventPredicate> correlationPredicates =
+            registration.correlationPredicates();
+        if (!correlationPredicates.isEmpty()) {
+          if (eventModel == null
+              && correlationPredicates.stream()
+                  .anyMatch(ModelAwareCloudEventPredicate.class::isInstance)) {
+            eventModel = registration.workflow().definition().application().modelFactory().from(ce);
+          }
+          if (!testCorrelation(ce, registration, eventModel)) {
+            continue;
+          }
+        }
+        registration.consumer().accept(ce);
+      }
+    }
+
+    private boolean testCorrelation(
+        CloudEvent ce, TypeEventRegistration registration, WorkflowModel eventModel) {
+      Collection<CloudEventPredicate> predicates = registration.correlationPredicates();
+      for (CloudEventPredicate pred : predicates) {
+        if (pred instanceof ModelAwareCloudEventPredicate ma) {
+          if (!ma.test(eventModel, registration.workflow(), registration.task())) {
+            return false;
+          }
+        } else {
+          if (!pred.test(ce, registration.workflow(), registration.task())) {
+            return false;
+          }
         }
       }
+      return true;
     }
 
     @Override
@@ -104,10 +160,16 @@ public TypeEventRegistration register(
       TaskContext task) {
     if (builder.type() == null) {
       registerToAll(ce);
-      return new TypeEventRegistration(null, ce, null, workflow, task);
+      return new TypeEventRegistration(null, ce, ALWAYS_TRUE, workflow, task);
     } else {
       TypeEventRegistration registration =
-          new TypeEventRegistration(builder.type(), ce, builder.cePredicate(), workflow, task);
+          new TypeEventRegistration(
+              builder.type(),
+              ce,
+              builder.cePredicate(),
+              builder.correlationPredicates(),
+              workflow,
+              task);
       registrations
           .computeIfAbsent(
               registration.type(),
@@ -138,5 +200,17 @@ public void unregister(TypeEventRegistration registration) {
             }
           });
     }
+    cleanupCorrelationState(registration);
+  }
+
+  private void cleanupCorrelationState(TypeEventRegistration registration) {
+    for (CloudEventPredicate pred : registration.correlationPredicates()) {
+      if (pred instanceof CorrelationPredicate cp) {
+        String key = cp.stateKey(registration.task());
+        if (key != null) {
+          registration.workflow().instance().removeMetadataIfPresent(key);
+        }
+      }
+    }
   }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/events/CorrelationPredicate.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.events;
+
+import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.api.types.CorrelateProperty;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CorrelationPredicate implements ModelAwareCloudEventPredicate {
+
+  private static final Logger logger = LoggerFactory.getLogger(CorrelationPredicate.class);
+
+  private final String correlationKey;
+  private final WorkflowValueResolver<Object> fromResolver;
+  private final WorkflowValueResolver<Object> expectResolver;
+
+  private CorrelationPredicate(
+      String correlationKey,
+      WorkflowValueResolver<Object> fromResolver,
+      WorkflowValueResolver<Object> expectResolver) {
+    this.correlationKey = correlationKey;
+    this.fromResolver = fromResolver;
+    this.expectResolver = expectResolver;
+  }
+
+  public static CorrelationPredicate from(
+      String key, CorrelateProperty prop, WorkflowApplication app) {
+    WorkflowValueResolver<Object> fromResolver =
+        app.expressionFactory().resolveValue(ExpressionDescriptor.from(prop.getFrom()));
+    WorkflowValueResolver<Object> expectResolver =
+        prop.getExpect() != null
+            ? app.expressionFactory().resolveValue(ExpressionDescriptor.from(prop.getExpect()))
+            : null;
+    return new CorrelationPredicate(key, fromResolver, expectResolver);
+  }
+
+  private String correlationStateKey(TaskContext task) {
+    return "correlation:"
+        + task.position().jsonPointer()
+        + ":"
+        + task.iteration()
+        + ":"
+        + correlationKey;
+  }
+
+  String stateKey(TaskContext task) {
+    return expectResolver == null ? correlationStateKey(task) : null;
+  }
+
+  @Override
+  public boolean test(CloudEvent cloudEvent, WorkflowContext workflow, TaskContext task) {
+    WorkflowModel eventModel = workflow.definition().application().modelFactory().from(cloudEvent);
+    return test(eventModel, workflow, task);
+  }
+
+  @Override
+  public boolean test(WorkflowModel eventModel, WorkflowContext workflow, TaskContext task) {
+    Object eventValue = fromResolver.apply(workflow, task, eventModel);
+    if (eventValue == null) {
+      logger.debug("Correlation from expression returned null");
+      return false;
+    }
+
+    if (expectResolver == null) {
+      String stateKey = correlationStateKey(task);
+      Object firstValue = workflow.instance().addMetadataIfAbsent(stateKey, () -> eventValue);
+      boolean result = Objects.equals(eventValue, firstValue);
+      logger.debug(
+          "Correlation no expect, eventValue='{}', firstValue='{}', match={}",
+          eventValue,
+          firstValue,
+          result);
+      return result;
+    }
+
+    Object expectedValue = expectResolver.apply(workflow, task, task.input());
+    boolean result = Objects.equals(eventValue, expectedValue);
+    logger.debug(
+        "Correlation eventValue='{}', expectedValue='{}', match={}",
+        eventValue,
+        expectedValue,
+        result);
+    return result;
+  }
+}