[Fix #1395] AllStrategyCorrelation persistence

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

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

@@ -16,6 +16,7 @@
 package io.serverlessworkflow.impl;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
 
 public interface WorkflowInstance extends WorkflowInstanceData {
   CompletableFuture<WorkflowModel> start();
@@ -49,4 +50,6 @@ public interface WorkflowInstance extends WorkflowInstanceData {
   boolean cancel();
 
   boolean resume();
+
+  <T> T addMetadataIfAbsent(String key, Supplier<T> supplier);
 }

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

@@ -16,6 +16,7 @@
 package io.serverlessworkflow.impl;
 
 import java.time.Instant;
+import java.util.Optional;
 
 public interface WorkflowInstanceData {
   String id();
@@ -29,4 +30,6 @@ public interface WorkflowInstanceData {
   WorkflowStatus status();
 
   WorkflowModel context();
+
+  <T> Optional<T> findMetadata(String key, Class<T> objectClass);
 }

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

@@ -41,6 +41,8 @@
 
 public class WorkflowMutableInstance implements WorkflowInstance {
 
+  public static final String CLOUD_EVENT_IDS = "CloudEventIds";
+
   private final AtomicReference<WorkflowStatus> status;
   protected final String id;
   protected final WorkflowModel input;
@@ -350,9 +352,13 @@ public void addCancelable(CompletableFuture<?> cancelable) {
     }
   }
 
-  public <T> T additionalObject(String key, Supplier<T> supplier) {
+  public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
   }
 
+  public <T> Optional<T> findMetadata(String key, Class<T> objectClass) {
+    return Optional.ofNullable(objectClass.cast(additionalObjects.get(key)));
+  }
+
   public void restoreContext(WorkflowContext workflow, TaskContext context) {}
 }

impl/core/src/main/java/io/serverlessworkflow/impl/scheduler/DefaultWorkflowScheduler.java

@@ -16,7 +16,7 @@
 package io.serverlessworkflow.impl.scheduler;
 
 import io.serverlessworkflow.impl.WorkflowDefinition;
-import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -83,10 +83,10 @@ protected CronResolverIntanceRunner(WorkflowDefinition definition) {
       }
 
       @Override
-      public void accept(WorkflowModel model) {
+      public void accept(WorkflowInstance instance) {
         if (!cancelled.get()) {
           scheduleNext();
-          super.accept(model);
+          super.accept(instance);
         }
       }
     }

impl/core/src/main/java/io/serverlessworkflow/impl/scheduler/ScheduledEventConsumer.java

@@ -19,8 +19,10 @@
 
 import io.cloudevents.CloudEvent;
 import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelCollection;
+import io.serverlessworkflow.impl.WorkflowMutableInstance;
 import io.serverlessworkflow.impl.events.EventConsumer;
 import io.serverlessworkflow.impl.events.EventRegistration;
 import io.serverlessworkflow.impl.events.EventRegistrationBuilderInfo;
@@ -78,13 +80,17 @@ public ScheduledEventConsumer(
   protected void start(CloudEvent ce) {
     WorkflowModelCollection model = definition.application().modelFactory().createCollection();
     model.add(converter.apply(ce));
-    instanceRunner.accept(model);
+    instanceRunner.accept(definition.instance(model));
   }
 
   protected void start(Collection<CloudEvent> ces) {
     WorkflowModelCollection model = definition.application().modelFactory().createCollection();
     ces.forEach(ce -> model.add(converter.apply(ce)));
-    instanceRunner.accept(model);
+    WorkflowInstance instance = definition.instance(model);
+    instance.addMetadataIfAbsent(
+        WorkflowMutableInstance.CLOUD_EVENT_IDS,
+        () -> ces.stream().map(CloudEvent::getId).toList());
+    instanceRunner.accept(instance);
   }
 
   public void close() {

impl/core/src/main/java/io/serverlessworkflow/impl/scheduler/ScheduledInstanceRunnable.java

@@ -20,7 +20,7 @@
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.util.function.Consumer;
 
-public class ScheduledInstanceRunnable implements Runnable, Consumer<WorkflowModel> {
+public class ScheduledInstanceRunnable implements Runnable, Consumer<WorkflowInstance> {
 
   protected final WorkflowDefinition definition;
 
@@ -30,16 +30,20 @@ public ScheduledInstanceRunnable(WorkflowDefinition definition) {
 
   @Override
   public void run() {
-    accept(definition.application().modelFactory().fromNull());
+    accept(definition.instance());
   }
 
   @Override
-  public void accept(WorkflowModel model) {
-    runScheduledInstance(definition, model);
+  public void accept(WorkflowInstance instance) {
+    runScheduledInstance(definition, instance);
   }
 
   public static void runScheduledInstance(WorkflowDefinition definition, WorkflowModel model) {
-    WorkflowInstance instance = definition.instance(model);
+    runScheduledInstance(definition, definition.instance(model));
+  }
+
+  private static void runScheduledInstance(
+      WorkflowDefinition definition, WorkflowInstance instance) {
     definition.addScheduledInstance(instance);
     definition.application().executorService().execute(() -> instance.start());
   }

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/AbstractAsyncPersistenceExecutor.java

@@ -15,16 +15,24 @@
  */
 package io.serverlessworkflow.impl.persistence;
 
-import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowDefinitionData;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 
 public abstract class AbstractAsyncPersistenceExecutor implements PersistenceExecutor {
   @Override
-  public CompletableFuture<Void> execute(Runnable runnable, WorkflowContextData context) {
+  public CompletableFuture<Void> execute(Runnable runnable, WorkflowDefinitionData definition) {
+
     return CompletableFuture.runAsync(
-        runnable, executorService().orElse(context.definition().application().executorService()));
+        runnable, executorService().orElse(definition.application().executorService()));
+  }
+
+  @Override
+  public <T> CompletableFuture<T> execute(Supplier<T> runnable, WorkflowDefinitionData definition) {
+    return CompletableFuture.supplyAsync(
+        runnable, executorService().orElse(definition.application().executorService()));
   }
 
   protected abstract Optional<ExecutorService> executorService();

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/AbstractPersistenceInstanceWriter.java

@@ -17,15 +17,25 @@
 
 import io.serverlessworkflow.impl.TaskContextData;
 import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowMutableInstance;
 import io.serverlessworkflow.impl.WorkflowStatus;
+import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 
 public abstract class AbstractPersistenceInstanceWriter implements PersistenceInstanceWriter {
 
   @Override
   public CompletableFuture<Void> started(WorkflowContextData workflowContext) {
-    return doStartInstance(t -> t.writeInstanceData(workflowContext), workflowContext);
+    return doStartInstance(
+        t -> {
+          t.writeInstanceData(workflowContext);
+          workflowContext
+              .instanceData()
+              .findMetadata(WorkflowMutableInstance.CLOUD_EVENT_IDS, Collection.class)
+              .ifPresent(c -> t.removeCloudEvents(c, workflowContext.definition()));
+        },
+        workflowContext);
   }
 
   @Override

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceAllStrategyCorrelationInfo.java

@@ -0,0 +1,52 @@
+/*
+ * 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.persistence;
+
+import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.events.EventRegistrationBuilder;
+import io.serverlessworkflow.impl.scheduler.AllStrategyCorrelationInfo;
+import java.util.Collection;
+import java.util.function.Consumer;
+
+public class PersistenceAllStrategyCorrelationInfo implements AllStrategyCorrelationInfo {
+
+  private final WorkflowDefinition definition;
+  private final PersistenceExecutor executor;
+  private final PersistenceInstanceOperations operations;
+
+  public PersistenceAllStrategyCorrelationInfo(
+      WorkflowDefinition definition,
+      PersistenceExecutor executor,
+      PersistenceInstanceOperations operations) {
+    this.definition = definition;
+    this.executor = executor;
+    this.operations = operations;
+  }
+
+  @Override
+  public void correlate(
+      EventRegistrationBuilder reg, CloudEvent event, Consumer<Collection<CloudEvent>> starter) {
+    executor
+        .execute(() -> operations.correlateEvent(event, definition), definition)
+        .thenAccept(events -> events.forEach(starter));
+  }
+
+  @Override
+  public void register(EventRegistrationBuilder reg) {
+    // empty operation
+  }
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceExecutor.java

@@ -15,11 +15,14 @@
  */
 package io.serverlessworkflow.impl.persistence;
 
-import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowDefinitionData;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
 
 public interface PersistenceExecutor extends AutoCloseable {
-  CompletableFuture<Void> execute(Runnable runnable, WorkflowContextData context);
+  <T> CompletableFuture<T> execute(Supplier<T> runnable, WorkflowDefinitionData definition);
+
+  CompletableFuture<Void> execute(Runnable runnable, WorkflowDefinitionData definition);
 
   default void close() {}
 }