feat: Implement correlation on event filter

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

@@ -37,8 +37,11 @@ public SELF with(Consumer<P> c) {
   }
 
   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");
+    ListenTaskBuilder.CorrelatePropertyBuilder cb =
+        new ListenTaskBuilder.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<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/events/AbstractTypeConsumer.java

@@ -16,12 +16,17 @@
 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 io.serverlessworkflow.impl.WorkflowModelFactory;
 import java.util.AbstractCollection;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
@@ -46,14 +51,34 @@ public abstract class AbstractTypeConsumer
   protected abstract void unregister(String topicName);
 
   private Map<String, CloudEventConsumer> registrations = new ConcurrentHashMap<>();
+  private WorkflowModelFactory modelFactory;
 
   @Override
   public TypeEventRegistrationBuilder listen(
       EventFilter register, WorkflowApplication application) {
+    this.modelFactory = application.modelFactory();
     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<CorrelationPredicate> correlationPredicates =
+        buildCorrelationPredicates(register.getCorrelate(), application);
+    return correlationPredicates.isEmpty()
+        ? new TypeEventRegistrationBuilder(type, cePredicate)
+        : new TypeEventRegistrationBuilder(type, cePredicate, correlationPredicates);
+  }
+
+  private Collection<CorrelationPredicate> buildCorrelationPredicates(
+      EventFilterCorrelate correlate, WorkflowApplication application) {
+    if (correlate == null || correlate.getAdditionalProperties().isEmpty()) {
+      return List.of();
+    }
+    Collection<CorrelationPredicate> predicates = new ArrayList<>();
+    for (Map.Entry<String, CorrelateProperty> entry :
+        correlate.getAdditionalProperties().entrySet()) {
+      predicates.add(CorrelationPredicate.from(entry.getKey(), entry.getValue(), application));
+    }
+    return predicates;
   }
 
   @Override
@@ -63,18 +88,45 @@ public Collection<TypeEventRegistrationBuilder> listenToAll(WorkflowApplication
 
   private static class CloudEventConsumer extends AbstractCollection<TypeEventRegistration>
       implements Consumer<CloudEvent> {
+    private final WorkflowModelFactory modelFactory;
     private Collection<TypeEventRegistration> registrations = new CopyOnWriteArrayList<>();
 
+    CloudEventConsumer(WorkflowModelFactory modelFactory) {
+      this.modelFactory = modelFactory;
+    }
+
     @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())) {
+          Collection<CorrelationPredicate> predicates = registration.correlationPredicates();
+          if (!predicates.isEmpty()) {
+            if (eventModel == null) {
+              eventModel = modelFactory.from(ce);
+            }
+            if (!testCorrelation(eventModel, registration)) {
+              continue;
+            }
+          }
           registration.consumer().accept(ce);
         }
       }
     }
 
+    private boolean testCorrelation(WorkflowModel eventModel, TypeEventRegistration registration) {
+      Collection<CorrelationPredicate> predicates = registration.correlationPredicates();
+      if (predicates.isEmpty()) {
+        return true;
+      }
+      for (CorrelationPredicate pred : predicates) {
+        if (!pred.test(eventModel, registration.workflow(), registration.task())) {
+          return false;
+        }
+      }
+      return true;
+    }
+
     @Override
     public boolean add(TypeEventRegistration registration) {
       return registrations.add(registration);
@@ -107,12 +159,18 @@ public TypeEventRegistration register(
       return new TypeEventRegistration(null, ce, null, 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(),
               k -> {
-                CloudEventConsumer consumer = new CloudEventConsumer();
+                CloudEventConsumer consumer = new CloudEventConsumer(modelFactory);
                 register(k, consumer);
                 return consumer;
               })

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

@@ -0,0 +1,85 @@
+/*
+ * 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.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;
+
+public class CorrelationPredicate {
+
+  private static final Logger logger = LoggerFactory.getLogger(CorrelationPredicate.class);
+
+  private final String key;
+  private final WorkflowValueResolver<Object> fromResolver;
+  private final WorkflowValueResolver<Object> expectResolver;
+
+  private CorrelationPredicate(
+      String key,
+      WorkflowValueResolver<Object> fromResolver,
+      WorkflowValueResolver<Object> expectResolver) {
+    this.key = key;
+    this.fromResolver = fromResolver;
+    this.expectResolver = expectResolver;
+  }
+
+  public static CorrelationPredicate from(
+      String key, io.serverlessworkflow.api.types.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);
+  }
+
+  public boolean test(WorkflowModel eventModel, WorkflowContext workflow, TaskContext task) {
+    Object eventValue = fromResolver.apply(workflow, task, eventModel);
+    if (eventValue == null) {
+      logger.debug("Correlation key '{}': from expression returned null", key);
+      return false;
+    }
+
+    if (expectResolver == null) {
+      logger.debug(
+          "Correlation key '{}': no expect expression, accepting event value '{}'",
+          key,
+          eventValue);
+      return true;
+    }
+
+    Object expectedValue = expectResolver.apply(workflow, task, task.input());
+    boolean result = Objects.equals(eventValue, expectedValue);
+    logger.debug(
+        "Correlation key '{}': eventValue='{}', expectedValue='{}', match={}",
+        key,
+        eventValue,
+        expectedValue,
+        result);
+    return result;
+  }
+
+  public String key() {
+    return key;
+  }
+}

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

@@ -18,12 +18,25 @@
 import io.cloudevents.CloudEvent;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.function.Consumer;
 
 public record TypeEventRegistration(
     String type,
     Consumer<CloudEvent> consumer,
     CloudEventPredicate predicate,
+    Collection<CorrelationPredicate> correlationPredicates,
     WorkflowContext workflow,
     TaskContext task)
-    implements EventRegistration {}
+    implements EventRegistration {
+
+  public TypeEventRegistration(
+      String type,
+      Consumer<CloudEvent> consumer,
+      CloudEventPredicate predicate,
+      WorkflowContext workflow,
+      TaskContext task) {
+    this(type, consumer, predicate, Collections.emptyList(), workflow, task);
+  }
+}

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

@@ -15,5 +15,16 @@
  */
 package io.serverlessworkflow.impl.events;
 
-public record TypeEventRegistrationBuilder(String type, CloudEventPredicate cePredicate)
-    implements EventRegistrationBuilder {}
+import java.util.Collection;
+import java.util.Collections;
+
+public record TypeEventRegistrationBuilder(
+    String type,
+    CloudEventPredicate cePredicate,
+    Collection<CorrelationPredicate> correlationPredicates)
+    implements EventRegistrationBuilder {
+
+  public TypeEventRegistrationBuilder(String type, CloudEventPredicate cePredicate) {
+    this(type, cePredicate, Collections.emptyList());
+  }
+}

impl/core/src/main/java/io/serverlessworkflow/impl/expressions/AbstractExpressionFactory.java

@@ -46,6 +46,12 @@ public WorkflowValueResolver<String> resolveString(ExpressionDescriptor desc) {
     return (w, t, m) -> toString(expr.eval(w, t, m));
   }
 
+  @Override
+  public WorkflowValueResolver<Object> resolveValue(ExpressionDescriptor desc) {
+    ObjectExpression expr = buildExpression(desc);
+    return (w, t, m) -> expr.eval(w, t, m);
+  }
+
   @Override
   public WorkflowValueResolver<Collection<?>> resolveCollection(ExpressionDescriptor desc) {
     ObjectExpression expr = buildExpression(desc);

impl/core/src/main/java/io/serverlessworkflow/impl/expressions/ExpressionFactory.java

@@ -38,6 +38,8 @@ default int priority(ExpressionDescriptor desc) {
 
   WorkflowValueResolver<String> resolveString(ExpressionDescriptor desc);
 
+  WorkflowValueResolver<Object> resolveValue(ExpressionDescriptor desc);
+
   WorkflowValueResolver<OffsetDateTime> resolveDate(ExpressionDescriptor desc);
 
   WorkflowValueResolver<CloudEventData> resolveCE(ExpressionDescriptor desc);