added async service

doc/RAG_SERVICE.md

@@ -1,4 +1,4 @@
-# The RAG ClusterService EJB
+# The RAG Cluster Service
 
 With the EJB `ClusterService` clients can index and retrieve embeddings.
 

doc/RAG_WORKFLOW.md

@@ -0,0 +1,63 @@
+# The RAG Workflow Integration
+
+The RAG system is integrated via the BPMN Model. There are several ways how the RAG information can be indexed, updated and deleted.
+
+## Indexing
+
+To index a process instance the Signal Adapter `ai.dynamixs.rag.RAGIndexAdapter` is used. The adapter expects a DataObject `PromptDefinition` associated with the corresponding event and a workflow configuration defining the LLM endpoint
+
+```xml
+<imixs-ai name="INDEX">
+  <endpoint>https://llama.cpp2.imixs.com/</endpoint>
+  <debug>true</debug>
+</imixs-ai>
+```
+
+The `debug` flag is optional and can be set to `true` to log index information.
+
+The PromptDefinition associated by a DataObject defines the text to generate the embeddings.
+
+```xml
+<?xml version="1.0" encoding="UTF-8"?>
+<PromptDefinition>
+  <prompt_options>{}</prompt_options>
+  <prompt><![CDATA[<itemvalue>$workflowgroup</itemvalue>: <itemvalue>$workflowstatus</itemvalue>
+<itemvalue>$workflowsummary</itemvalue>
+
+# Summary
+<itemvalue>$workflowabstract</itemvalue>
+
+]]>
+    </prompt>
+</PromptDefinition>
+```
+
+The RAG System does not only store the embeddings but also the following Workflow Meta Information
+
+- ModelVersion - the `$modelversion` item assigned to the process instance at the time of the indexing
+- TaskId - the `$taskId` item assigned to the process instance at the time of the indexing
+
+**Note:** The workflow meta information is generated at the time of the indexing and does not necessary match the current workflow status!
+
+## Update
+
+To update the workflow metadata only, plugin class `ai.dynamixs.rag.RAGIndexPlugin` can be used.
+This plugin is used to update the workflow model version and the workflow status only but not the embedings. The update is only be performed if an index already exists.
+
+The update is performed automatically in each workflow processing cycle. To disabled a metadata update in single specific event you can define the following workflow definition:
+
+```xml
+<rag-index name="DISABLE"></rag-index>
+```
+
+In this case the plugin will be disabled.
+
+## Delete
+
+To delete a RAG index the plugin class `ai.dynamixs.rag.RAGPlugin` can be used in the DELETE mode:
+
+```xml
+<rag-index name="DELETE"></rag-index>
+```
+
+This will remove the index for the current workitem.

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGAdapter.java

@@ -42,6 +42,9 @@ import org.imixs.workflow.engine.WorkflowService;
 import org.imixs.workflow.exceptions.AdapterException;
 import org.imixs.workflow.exceptions.PluginException;
 
+import ai.dynamixs.rag.cluster.ClusterException;
+import ai.dynamixs.rag.cluster.ClusterService;
+import ai.dynamixs.rag.cluster.RetrievalResult;
 import jakarta.inject.Inject;
 
 /**

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGPlugin.java

@@ -0,0 +1,107 @@
+/*  
+ *  Imixs-Workflow 
+ *  
+ *  Copyright (C) 2001-2020 Imixs Software Solutions GmbH,  
+ *  http://www.imixs.com
+ *  
+ *  This program is free software; you can redistribute it and/or 
+ *  modify it under the terms of the GNU General Public License 
+ *  as published by the Free Software Foundation; either version 2 
+ *  of the License, or (at your option) any later version.
+ *  
+ *  This program is distributed in the hope that it will be useful, 
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of 
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 
+ *  General Public License for more details.
+ *  
+ *  You can receive a copy of the GNU General Public
+ *  License at http://www.gnu.org/licenses/gpl.html
+ *  
+ *  Project: 
+ *      https://www.imixs.org
+ *      https://github.com/imixs/imixs-workflow
+ *  
+ *  Contributors:  
+ *      Imixs Software Solutions GmbH - Project Management
+ *      Ralph Soika - Software Developer
+ */
+
+package ai.dynamixs.rag;
+
+import java.util.List;
+import java.util.logging.Logger;
+
+import org.imixs.workflow.ItemCollection;
+import org.imixs.workflow.engine.EventLogService;
+import org.imixs.workflow.engine.WorkflowService;
+import org.imixs.workflow.engine.plugins.AbstractPlugin;
+import org.imixs.workflow.exceptions.PluginException;
+
+import ai.dynamixs.rag.events.RAGEventService;
+import jakarta.inject.Inject;
+
+/**
+ * The RAGIndexPlugin updates the workflow meta data for existing embeddings.
+ * The plugin can be deactivated with:
+ * <p>
+ * 
+ * <pre>
+  {@code
+   <rag-index name="DISABLE">
+      <debug>false</debug>  
+   </rag-index>
+    }
+ * </pre>
+ * <p>
+ * The tag 'debug' is optional
+ * 
+ * 
+ * @author rsoika
+ * 
+ */
+public class RAGPlugin extends AbstractPlugin {
+
+	private static final Logger logger = Logger.getLogger(RAGPlugin.class.getName());
+
+	@Inject
+	private WorkflowService workflowService;
+
+	@Inject
+	EventLogService eventLogService;
+
+	@Override
+	public ItemCollection run(ItemCollection workitem, ItemCollection event) throws PluginException {
+		List<ItemCollection> deleteDefinitions = null;
+		List<ItemCollection> disableDefinitions = null;
+
+		logger.finest("running TaxonomyPlugin");
+		String workflowResult = event.getItemValueString("workflow.result");
+		if (workflowResult.contains("<rag-index")) {
+			deleteDefinitions = workflowService.evalWorkflowResultXML(event, "rag-index",
+					"DELETE", workitem, false);
+
+			disableDefinitions = workflowService.evalWorkflowResultXML(event,
+					"rag-index",
+					"DISABLE", workitem, false);
+		}
+
+		// disabled?
+		if (disableDefinitions != null && disableDefinitions.size() > 0) {
+			// no op!
+			return workitem;
+		}
+
+		// delete mode?
+		if (deleteDefinitions != null && deleteDefinitions.size() > 0) {
+			eventLogService.createEvent(RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE, workitem.getUniqueID());
+			return workitem;
+		}
+
+		// default update meta information
+		eventLogService.createEvent(RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE, workitem.getUniqueID());
+
+		return workitem;
+
+	}
+
+}

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterException.java

@@ -25,7 +25,7 @@
  *  	Ralph Soika - Software Developer
  *******************************************************************************/
 
-package ai.dynamixs.rag;
+package ai.dynamixs.rag.cluster;
 
 /**
  * The ClusterService provides methods to persist the content of a Imixs AI

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java

@@ -1,4 +1,4 @@
-package ai.dynamixs.rag;
+package ai.dynamixs.rag.cluster;
 
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
@@ -137,6 +137,8 @@ public class ClusterService {
     private final Map<String, PreparedStatement> searchVectorStatements = new HashMap<>();
     private PreparedStatement selectVectorStmt;
     private PreparedStatement removeVectorStmt;
+    private PreparedStatement selectChunksStmt = null;
+    private PreparedStatement updateChunkStmt = null;
 
     @Resource
     private TimerService timerService;
@@ -414,20 +416,10 @@ public class ClusterService {
             String deleteQuery = "DELETE FROM document_vectors WHERE id = ?";
             removeVectorStmt = session.prepare(deleteQuery);
         }
-        if (selectVectorStmt == null) {
-            String selectQuery = "SELECT COUNT(*) FROM document_vectors WHERE id = ?";
-            selectVectorStmt = session.prepare(selectQuery);
-        }
 
         try {
             // test if entries exist
-            BoundStatement selectBoundStmt = selectVectorStmt.bind(uniqueID);
+            long count = countIndexEntries(uniqueID);
-            ResultSet resultSet = session.execute(selectBoundStmt);
-            Row row = resultSet.one();
-
-            if (row != null) {
-                long count = row.getLong(0);
-
             if (count > 0) {
                 // remove all
                 BoundStatement deleteBoundStmt = removeVectorStmt.bind(uniqueID);
@@ -437,9 +429,6 @@ public class ClusterService {
             } else {
                 // logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueID);
             }
-            } else {
-                // logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueID);
-            }
 
         } catch (Exception e) {
             throw new ClusterException(ClusterException.CLUSTER_ERROR,
@@ -448,6 +437,36 @@ public class ClusterService {
         return false;
     }
 
+    /**
+     * This method returns the number of index entries with the given uniqueId. The
+     * method can be used to test if index data is available.
+     * 
+     * @param uniqueId
+     * @return - count of entries, or 0 if no entries exist
+     * @throws ClusterException
+     */
+    public long countIndexEntries(String uniqueId) throws ClusterException {
+        if (selectVectorStmt == null) {
+            String selectQuery = "SELECT COUNT(*) FROM document_vectors WHERE id = ?";
+            selectVectorStmt = session.prepare(selectQuery);
+        }
+        try {
+            // test if entries exist
+            BoundStatement selectBoundStmt = selectVectorStmt.bind(uniqueId);
+            ResultSet resultSet = session.execute(selectBoundStmt);
+            Row row = resultSet.one();
+
+            if (row != null) {
+                long count = row.getLong(0);
+                return count;
+            }
+        } catch (Exception e) {
+            throw new ClusterException(ClusterException.CLUSTER_ERROR,
+                    "Failed to lookup entries for uniqueID '" + uniqueId + "': " + e.getMessage(), e);
+        }
+        return 0;
+    }
+
     /**
      * Performs a semantic search on the document vector database using cosine
      * similarity.
@@ -632,6 +651,63 @@ public class ClusterService {
         return searchEmbeddings(embedding, maxResults, Arrays.asList(modelPattern), taskIds, null, null);
     }
 
+    /**
+     * Update the workflow meta data. The method returns true if a index with the
+     * given uniqueId was found
+     * 
+     * @param uniqueId
+     * @param modelVersion
+     * @param taskId
+     * @return true if an update was performed
+     * @throws ClusterException
+     */
+    public boolean updateMetaData(String uniqueId, String modelVersion, int taskId)
+            throws ClusterException {
+
+        // Prepare statements if not already done
+        if (selectChunksStmt == null) {
+            String selectQuery = "SELECT chunk_id FROM document_vectors WHERE id = ?";
+            selectChunksStmt = session.prepare(selectQuery);
+        }
+
+        if (updateChunkStmt == null) {
+            String updateQuery = "UPDATE document_vectors SET model_version = ?, task_id = ? WHERE id = ? AND chunk_id = ?";
+            updateChunkStmt = session.prepare(updateQuery);
+        }
+        try {
+            // 1. Alle chunk_ids für die uniqueId ermitteln
+            BoundStatement selectBoundStmt = selectChunksStmt.bind(uniqueId);
+            ResultSet resultSet = session.execute(selectBoundStmt);
+
+            List<UUID> chunkIds = new ArrayList<>();
+            for (Row row : resultSet) {
+                chunkIds.add(row.getUuid("chunk_id"));
+            }
+
+            if (chunkIds.isEmpty()) {
+                logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueId);
+                return false;
+            }
+
+            // 2. Jede gefundene chunk_id einzeln updaten
+            int updateCount = 0;
+            for (UUID chunkId : chunkIds) {
+                BoundStatement updateBoundStmt = updateChunkStmt.bind(modelVersion, taskId, uniqueId, chunkId);
+                session.execute(updateBoundStmt);
+                updateCount++;
+            }
+
+            logger.info("│ ├── Updated metadata for " + updateCount + " entries with uniqueID: " +
+                    uniqueId + " (model: " + modelVersion + ", task: " + taskId + ")");
+            return true;
+
+        } catch (Exception e) {
+            throw new ClusterException(ClusterException.CLUSTER_ERROR,
+                    "Failed to update metadata for uniqueID '" + uniqueId + "': " + e.getMessage(), e);
+        }
+
+    }
+
     /**
      * Helper method to prepare a IN clause
      * 

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/RetrievalResult.java

@@ -1,4 +1,4 @@
-package ai.dynamixs.rag;
+package ai.dynamixs.rag.cluster;
 
 /**
  * Represents a single result from a semantic search operation in the RAG

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventScheduler.java

@@ -0,0 +1,145 @@
+/*  
+ *  Imixs-Workflow 
+ *  
+ *  Copyright (C) 2001-2020 Imixs Software Solutions GmbH,  
+ *  http://www.imixs.com
+ *  
+ *  This program is free software; you can redistribute it and/or 
+ *  modify it under the terms of the GNU General Public License 
+ *  as published by the Free Software Foundation; either version 2 
+ *  of the License, or (at your option) any later version.
+ *  
+ *  This program is distributed in the hope that it will be useful, 
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of 
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 
+ *  General Public License for more details.
+ *  
+ *  You can receive a copy of the GNU General Public
+ *  License at http://www.gnu.org/licenses/gpl.html
+ *  
+ *  Project: 
+ *      https://www.imixs.org
+ *      https://github.com/imixs/imixs-workflow
+ *  
+ *  Contributors:  
+ *      Imixs Software Solutions GmbH - Project Management
+ *      Ralph Soika - Software Developer
+ */
+
+package ai.dynamixs.rag.events;
+
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_DEADLOCK;
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_ENABLED;
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_INITIALDELAY;
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_INTERVAL;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.imixs.workflow.engine.AsyncEventService;
+import org.imixs.workflow.engine.EventLogService;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import jakarta.annotation.security.DeclareRoles;
+import jakarta.annotation.security.RunAs;
+import jakarta.ejb.Singleton;
+import jakarta.ejb.Startup;
+import jakarta.ejb.Timeout;
+import jakarta.ejb.Timer;
+import jakarta.ejb.TimerConfig;
+import jakarta.ejb.TimerService;
+import jakarta.inject.Inject;
+
+/**
+ * The RAGEventScheduler starts a scheduler service to process RAG events in an
+ * asynchronous way by calling the RAGEventService.
+ * <p>
+ * The AsyncEventScheduler runs on a non-persistent ejb timer with the same
+ * interval settings as the AsyncEventProcessor
+ * <p>
+ * RAGEventScheduler 'ASYNCEVENT_PROCESSOR_INTERVAL' and an optional delay
+ * defined by 'ASYNCEVENT_PROCESSOR_INITIALDELAY'. To enable the processor
+ * 'ASYNCEVENT_PROCESSOR_ENABLED' must be set to true (default=false).
+ * 'ASYNCEVENT_PROCESSOR_DEADLOCK' deadlock timeout
+ * <p>
+ * In a clustered environment this timer runs in each cluster member that
+ * contains the EJB. So this means the non-persistent EJB Timer scales
+ * horizontal within a clustered environment – e.g. a Kubernetes cluster.
+ *
+ * @see AsyncEventService
+ * @version 1.1
+ * @author rsoika
+ *
+ */
+@DeclareRoles({ "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
+@RunAs("org.imixs.ACCESSLEVEL.MANAGERACCESS")
+@Startup
+@Singleton
+public class RAGEventScheduler {
+
+    // enabled
+    @Inject
+    @ConfigProperty(name = ASYNCEVENT_PROCESSOR_ENABLED, defaultValue = "false")
+    boolean enabled;
+
+    // timeout interval in ms
+    @Inject
+    @ConfigProperty(name = ASYNCEVENT_PROCESSOR_INTERVAL, defaultValue = "1000")
+    long interval;
+
+    // initial delay in ms
+    @Inject
+    @ConfigProperty(name = ASYNCEVENT_PROCESSOR_INITIALDELAY, defaultValue = "0")
+    long initialDelay;
+
+    // deadlock timeout interval in ms
+    @Inject
+    @ConfigProperty(name = ASYNCEVENT_PROCESSOR_DEADLOCK, defaultValue = "60000")
+    long deadLockInterval;
+
+    private static final Logger logger = Logger.getLogger(RAGEventScheduler.class.getName());
+
+    @Resource
+    TimerService timerService;
+
+    @Inject
+    RAGEventService ragEventService;
+
+    @Inject
+    EventLogService eventLogService;
+
+    @PostConstruct
+    public void init() {
+        if (enabled) {
+
+            // Registering a non-persistent Timer Service.
+            final TimerConfig timerConfig = new TimerConfig();
+            timerConfig.setInfo("Imixs-Workflow AsyncEventScheduler");
+            timerConfig.setPersistent(false);
+            timerService.createIntervalTimer(initialDelay, interval, timerConfig);
+            logger.log(Level.INFO, "├── ✅ Started RAGEventScheduler - initalDelay={0}  inverval={1}",
+                    new Object[] { initialDelay, interval });
+
+        }
+    }
+
+    /**
+     * The method delegates the event processing to the stateless ejb
+     * AsyncEventProcessor.
+     * <p>
+     * Before processing the eventLog the method releases possible dead locks first.
+     * Both methods are running in separate transactions
+     * 
+     */
+    @Timeout
+    public void run(Timer timer) {
+        eventLogService.releaseDeadLocks(deadLockInterval,
+                RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX,
+                RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE,
+                RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE);
+        ragEventService.processEventLog();
+    }
+
+}

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventService.java

@@ -0,0 +1,173 @@
+/*  
+ *  Imixs-Workflow 
+ *  
+ *  Copyright (C) 2001-2020 Imixs Software Solutions GmbH,  
+ *  http://www.imixs.com
+ *  
+ *  This program is free software; you can redistribute it and/or 
+ *  modify it under the terms of the GNU General Public License 
+ *  as published by the Free Software Foundation; either version 2 
+ *  of the License, or (at your option) any later version.
+ *  
+ *  This program is distributed in the hope that it will be useful, 
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of 
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 
+ *  General Public License for more details.
+ *  
+ *  You can receive a copy of the GNU General Public
+ *  License at http://www.gnu.org/licenses/gpl.html
+ *  
+ *  Project: 
+ *      https://www.imixs.org
+ *      https://github.com/imixs/imixs-workflow
+ *  
+ *  Contributors:  
+ *      Imixs Software Solutions GmbH - Project Management
+ *      Ralph Soika - Software Developer
+ */
+
+package ai.dynamixs.rag.events;
+
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.commons.compress.archivers.ArchiveException;
+import org.imixs.workflow.ItemCollection;
+import org.imixs.workflow.engine.AsyncEventScheduler;
+import org.imixs.workflow.engine.EventLogService;
+import org.imixs.workflow.engine.WorkflowService;
+import org.imixs.workflow.engine.jpa.EventLog;
+
+import ai.dynamixs.rag.cluster.ClusterException;
+import ai.dynamixs.rag.cluster.ClusterService;
+import jakarta.annotation.security.DeclareRoles;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.ejb.Stateless;
+import jakarta.ejb.TransactionAttribute;
+import jakarta.ejb.TransactionAttributeType;
+import jakarta.inject.Inject;
+import jakarta.persistence.OptimisticLockException;
+
+/**
+ * The AsyncEventService can be used to process workflow events in an
+ * asynchronous batch process. The AsyncEventService lookup eventLog entries of
+ * the topic "async.event".
+ * <p>
+ * The processor look up the workItem and starts a processing life cycle.
+ * <p>
+ * The AsyncEventService is called only by the AsyncEventScheduler which is
+ * implementing a ManagedScheduledExecutorService.
+ * <p>
+ * To prevent concurrent processes to handle the same workitems the batch
+ * process uses a Optimistic lock strategy. After fetching new event log entries
+ * the processor updates the eventLog entry in a new transaction and set the
+ * topic to 'batch.process.lock'. After that update we can be sure that no other
+ * process is dealing with these entries. After completing the processing step
+ * the eventlog entry will be removed.
+ * <p>
+ * To avoid ad deadlock the processor set an expiration time on the lock, so the
+ * lock will be auto-released after 1 minute (batch.processor.deadlock).
+ * 
+ * @see AsyncEventScheduler
+ * @version 1.0
+ * @author rsoika
+ *
+ */
+@DeclareRoles({ "org.imixs.ACCESSLEVEL.NOACCESS", "org.imixs.ACCESSLEVEL.READERACCESS",
+        "org.imixs.ACCESSLEVEL.AUTHORACCESS", "org.imixs.ACCESSLEVEL.EDITORACCESS",
+        "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
+@RolesAllowed({ "org.imixs.ACCESSLEVEL.NOACCESS", "org.imixs.ACCESSLEVEL.READERACCESS",
+        "org.imixs.ACCESSLEVEL.AUTHORACCESS", "org.imixs.ACCESSLEVEL.EDITORACCESS",
+        "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
+@Stateless
+public class RAGEventService {
+
+    public static final String EVENTLOG_TOPIC_RAG_EVENT_INDEX = "rag.event.index";
+    public static final String EVENTLOG_TOPIC_RAG_EVENT_DELETE = "rag.event.delete";
+    public static final String EVENTLOG_TOPIC_RAG_EVENT_UPDATE = "rag.event.update";
+
+    private static final Logger logger = Logger.getLogger(RAGEventService.class.getName());
+
+    @Inject
+    EventLogService eventLogService;
+
+    @Inject
+    private WorkflowService workflowService;
+
+    @Inject
+    private ClusterService clusterService;
+
+    /**
+     * The method lookups for RAG event log entries and updates the RAG index
+     * information
+     * <p>
+     * Each eventLogEntry provides optional a prompt template
+     * 
+     * @throws ArchiveException
+     */
+    @TransactionAttribute(value = TransactionAttributeType.REQUIRES_NEW)
+    public void processEventLog() {
+        long l = System.currentTimeMillis();
+        boolean debug = logger.isLoggable(Level.FINE);
+
+        // test for new event log entries by timeout...
+        List<EventLog> events = eventLogService.findEventsByTimeout(100,
+                RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX,
+                RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE,
+                RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE);
+
+        if (events.size() > 0) {
+            logger.log(Level.INFO, "├── 🔃 processing {0} RAG events....", events.size());
+
+            for (EventLog eventLogEntry : events) {
+                try {
+
+                    // first try to lock the eventLog entry....
+                    if (eventLogService.lock(eventLogEntry)) {
+                        ItemCollection workitem = workflowService.getWorkItem(eventLogEntry.getRef());
+                        if (workitem == null) {
+                            logger.log(Level.WARNING, "│   ├── ⚠️ unable to read workitem: " + eventLogEntry.getRef());
+                            eventLogService.removeEvent(eventLogEntry.getId());
+                            continue;
+                        }
+                        switch (eventLogEntry.getTopic()) {
+                        case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE + ".lock":
+                            // remove embeddings
+                            clusterService.removeAllEmbeddings(eventLogEntry.getRef());
+                            break;
+
+                        case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE + ".lock":
+                            // update workflow status
+                            clusterService.updateMetaData(eventLogEntry.getRef(), workitem.getModelVersion(),
+                                    workitem.getTaskID());
+                            break;
+                        case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX:
+                            logger.log(Level.WARNING, "│   ├── ⚠️ RAG_INDEX Event is not yet supported!");
+                            break;
+                        default:
+                            // no op
+                        }
+
+                        // finally remove the event log entry...
+                        eventLogService.removeEvent(eventLogEntry.getId());
+                    }
+
+                } catch (OptimisticLockException e) {
+                    // lock was not possible - continue....
+                    logger.log(Level.INFO, "...unable to lock AsyncEvent: {0}", e.getMessage());
+                } catch (ClusterException e) {
+                    logger.log(Level.WARNING, e.getErrorCode() + " - " + e.getMessage());
+                    // remove the event log entry...
+                    eventLogService.removeEvent(eventLogEntry.getId());
+                }
+
+            }
+
+            logger.log(Level.INFO, "├── ✅ {0} RAGEvents processed in {1}ms",
+                    new Object[] { events.size(), System.currentTimeMillis() - l });
+
+        }
+    }
+
+}

workflow/product-de-1.0.0.bpmn

@@ -34,6 +34,7 @@
       <imixs:value><![CDATA[org.imixs.marty.profile.MailPlugin]]></imixs:value>
       <imixs:value><![CDATA[org.imixs.workflow.engine.plugins.IntervalPlugin]]></imixs:value>
       <imixs:value><![CDATA[org.imixs.workflow.engine.plugins.SplitAndJoinPlugin]]></imixs:value>
+      <imixs:value><![CDATA[ai.dynamixs.rag.RAGPlugin]]></imixs:value>
     </imixs:item>
     <open-bpmn:auto-align>true</open-bpmn:auto-align>
   </bpmn2:extensionElements>