[Fix #1691] MVStore additional objects persistence (#1715)

* [Fix #1691] Add metadata to mvstore

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

* [Fix #1691] MapInMemory approach

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

* [Fix #1691] Moving from int index to string generated index

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

* [Fix #1691] Use Ulid for improving versatility

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

---------

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

Author: fjtirado

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

@@ -53,7 +53,7 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   protected AtomicReference<CompletableFuture<WorkflowModel>> futureRef = new AtomicReference<>();
   protected Instant completedAt;
 
-  protected final Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
+  private Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
 
   protected final Map<String, Integer> iterationsMap = new ConcurrentHashMap<>();
 
@@ -448,6 +448,10 @@ public void addCancelable(CompletableFuture<?> cancelable) {
     }
   }
 
+  protected void setMetadata(Map<String, Object> metadata) {
+    this.additionalObjects = new ConcurrentHashMap<>(metadata);
+  }
+
   @Override
   public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultOutputBuffer.java

@@ -122,6 +122,16 @@ public WorkflowOutputBuffer writeBytes(byte[] bytes) {
     return this;
   }
 
+  @Override
+  public WorkflowOutputBuffer writeRawBytes(byte[] bytes) {
+    try {
+      output.write(bytes);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+    return this;
+  }
+
   @Override
   public void close() {
     try {

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java

@@ -173,6 +173,10 @@ public static WorkflowModel readModel(WorkflowBufferFactory factory, byte[] valu
     return readValue(factory, value, b -> (WorkflowModel) b.readObject());
   }
 
+  public static Object readObject(WorkflowBufferFactory factory, byte[] value) {
+    return readValue(factory, value, b -> b.readObject());
+  }
+
   public static Instant readInstant(WorkflowBufferFactory factory, byte[] value) {
     return readValue(factory, value, WorkflowInputBuffer::readInstant);
   }

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowOutputBuffer.java

@@ -41,6 +41,8 @@ public interface WorkflowOutputBuffer extends AutoCloseable {
 
   WorkflowOutputBuffer writeBytes(byte[] bytes);
 
+  WorkflowOutputBuffer writeRawBytes(byte[] bytes);
+
   WorkflowOutputBuffer writeInstant(Instant instant);
 
   default WorkflowOutputBuffer writeOffsetDateTime(OffsetDateTime time) {

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

@@ -17,14 +17,16 @@
 
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.time.Instant;
+import java.util.Map;
 
 public record CompletedTaskInfo(
     Instant instant,
     WorkflowModel model,
     WorkflowModel context,
     Boolean isEndNode,
     String nextPosition,
-    int iteration)
+    int iteration,
+    Map<String, Object> additionalObjects)
     implements PersistenceTaskInfo {
 
   public CompletedTaskInfo(
@@ -35,4 +37,14 @@ public CompletedTaskInfo(
       String nextPosition) {
     this(instant, model, context, isEndNode, nextPosition, 1);
   }
+
+  public CompletedTaskInfo(
+      Instant instant,
+      WorkflowModel model,
+      WorkflowModel context,
+      Boolean isEndNode,
+      String nextPosition,
+      int iteration) {
+    this(instant, model, context, isEndNode, nextPosition, iteration, Map.of());
+  }
 }

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

@@ -17,5 +17,12 @@
 
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.time.Instant;
+import java.util.Map;
 
-public record PersistenceInstanceInfo(Instant startedAt, WorkflowModel input) {}
+public record PersistenceInstanceInfo(
+    Instant startedAt, WorkflowModel input, Map<String, Object> metadata) {
+
+  public PersistenceInstanceInfo(Instant startedAt, WorkflowModel input) {
+    this(startedAt, input, Map.of());
+  }
+}

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

@@ -25,4 +25,15 @@ public record PersistenceWorkflowInfo(
     Instant startedAt,
     WorkflowModel input,
     WorkflowStatus status,
-    Map<String, PersistenceTaskInfo> tasks) {}
+    Map<String, PersistenceTaskInfo> tasks,
+    Map<String, Object> metadata) {
+
+  public PersistenceWorkflowInfo(
+      String id,
+      Instant startedAt,
+      WorkflowModel input,
+      WorkflowStatus status,
+      Map<String, PersistenceTaskInfo> tasks) {
+    this(id, startedAt, input, status, tasks, Map.of());
+  }
+}

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

@@ -15,4 +15,11 @@
  */
 package io.serverlessworkflow.impl.persistence;
 
-public record RetriedTaskInfo(int retryAttempt) implements PersistenceTaskInfo {}
+import java.util.Map;
+
+public record RetriedTaskInfo(int retryAttempt, Map<String, Object> metadata)
+    implements PersistenceTaskInfo {
+  public RetriedTaskInfo(int retryAttempt) {
+    this(retryAttempt, Map.of());
+  }
+}

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

@@ -48,6 +48,7 @@ private WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWo
               }
             });
     this.startedAt = info.startedAt();
+    setMetadata(info.metadata());
   }
 
   @Override
@@ -83,6 +84,7 @@ public void restoreContext(WorkflowContext workflow, TaskContext context) {
                   : workflow.definition().taskExecutor(completedTaskInfo.nextPosition()),
               completedTaskInfo.isEndNode()));
       workflow.context(completedTaskInfo.context());
+      setMetadata(completedTaskInfo.additionalObjects());
     } else if (taskInfo instanceof RetriedTaskInfo retriedTaskInfo) {
       if (context.retryAttempt() == 0) {
         context.retryAttempt(retriedTaskInfo.retryAttempt());
@@ -98,6 +100,7 @@ public void restoreContext(WorkflowContext workflow, TaskContext context) {
         }
         searchContext = tryContext.parent();
       }
+      setMetadata(retriedTaskInfo.metadata());
     }
   }
 }

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java

@@ -0,0 +1,71 @@
+/*
+ * 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.hashing;
+
+import com.github.f4b6a3.ulid.Ulid;
+import com.github.f4b6a3.ulid.UlidFactory;
+import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
+import java.util.Optional;
+
+public class DefaultHashFactory implements HashFactory {
+
+  private final UlidFactory idFactory = UlidFactory.newMonotonicInstance();
+
+  @Override
+  public Optional<HashItem> fromBuffer(byte id, WorkflowInputBuffer buffer) {
+    return Optional.ofNullable(
+        switch (id) {
+          case MD5HashItem.ID -> new MD5HashItem(buffer);
+          case IntegerHashItem.ID -> new IntegerHashItem(buffer);
+          case HashItem.HASHING_DISABLED -> null;
+          default -> throw new UnsupportedOperationException("Unsupported id " + id);
+        });
+  }
+
+  @Override
+  public Optional<HashItem> fromData(byte[] data) {
+    if (md5Condition(data)) {
+      return Optional.of(new MD5HashItem(data));
+    } else if (intCondition(data)) {
+      return Optional.of(new IntegerHashItem(data));
+    } else {
+      return Optional.empty();
+    }
+  }
+
+  protected boolean intCondition(byte[] data) {
+    return data.length > IntegerHashItem.SIZE_THRESHOLD;
+  }
+
+  protected boolean md5Condition(byte[] data) {
+    return data.length > MD5HashItem.SIZE_THRESHOLD;
+  }
+
+  @Override
+  public HashIndex indexFromBytes(byte[] bytes) {
+    return new DefaultHashIndex(Ulid.from(bytes));
+  }
+
+  @Override
+  public HashIndex indexFromString(String str) {
+    return new DefaultHashIndex(Ulid.from(str));
+  }
+
+  @Override
+  public HashIndex newIndex() {
+    return new DefaultHashIndex(idFactory.create());
+  }
+}