feat: replace process definition graphs when no instance is running

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-11 16:15:17 +08:00
parent a6410b3dc4
commit 946c9d3c52
16 changed files with 277 additions and 48 deletions
@@ -30,6 +30,14 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
            "INSERT INTO ordo_step_candidate (definition_id, step_id, candidate, candidate_order) VALUES (?, ?, ?, ?)";
    private static final String INSERT_TRANSITION =
            "INSERT INTO ordo_step_transition (definition_id, from_step_id, to_step_id, condition_key, priority) VALUES (?, ?, ?, ?, ?)";
+    private static final String UPDATE_DEFINITION_NAME =
+            "UPDATE ordo_process_definition SET name = ? WHERE id = ?";
+    private static final String DELETE_TRANSITIONS =
+            "DELETE FROM ordo_step_transition WHERE definition_id = ?";
+    private static final String DELETE_CANDIDATES =
+            "DELETE FROM ordo_step_candidate WHERE definition_id = ?";
+    private static final String DELETE_STEPS =
+            "DELETE FROM ordo_approval_step WHERE definition_id = ?";
    private static final String SELECT_DEFINITION =
            "SELECT id, name FROM ordo_process_definition WHERE id = ?";
    private static final String SELECT_STEPS =
@@ -51,47 +59,10 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
        Objects.requireNonNull(definition, "definition must not be null");
        Connection connection = connectionProvider.getConnection();
        try {
-            try (PreparedStatement insertDefinition = connection.prepareStatement(INSERT_DEFINITION)) {
-                insertDefinition.setString(1, definition.id());
-                insertDefinition.setString(2, definition.name());
-                insertDefinition.executeUpdate();
-            } catch (SQLException e) {
-                if (isDuplicateKey(e)) {
-                    return false;
-                }
-                throw new JdbcStorageException("failed to insert definition: " + definition.id(), e);
-            }
-            int stepOrder = 0;
-            for (ApprovalStep step : definition.steps()) {
-                try (PreparedStatement insertStep = connection.prepareStatement(INSERT_STEP)) {
-                    insertStep.setString(1, definition.id());
-                    insertStep.setString(2, step.id());
-                    insertStep.setString(3, step.name());
-                    insertStep.setString(4, step.policy().name());
-                    insertStep.setInt(5, stepOrder++);
-                    insertStep.executeUpdate();
-                }
-                int candidateOrder = 0;
-                for (String candidate : step.candidates()) {
-                    try (PreparedStatement insertCandidate = connection.prepareStatement(INSERT_CANDIDATE)) {
-                        insertCandidate.setString(1, definition.id());
-                        insertCandidate.setString(2, step.id());
-                        insertCandidate.setString(3, candidate);
-                        insertCandidate.setInt(4, candidateOrder++);
-                        insertCandidate.executeUpdate();
-                    }
-                }
-            }
-            for (StepTransition transition : definition.transitions()) {
-                try (PreparedStatement insertTransition = connection.prepareStatement(INSERT_TRANSITION)) {
-                    insertTransition.setString(1, definition.id());
-                    insertTransition.setString(2, transition.fromStepId());
-                    insertTransition.setString(3, transition.toStepId());
-                    insertTransition.setString(4, transition.conditionKey());
-                    insertTransition.setInt(5, transition.priority());
-                    insertTransition.executeUpdate();
-                }
+            if (!insertDefinitionRow(connection, definition)) {
+                return false;
            }
+            insertGraph(connection, definition);
            return true;
        } catch (SQLException e) {
            throw new JdbcStorageException("failed to insert definition: " + definition.id(), e);
@@ -100,6 +71,23 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
        }
    }

+    @Override
+    public void upsert(ProcessDefinition definition) {
+        Objects.requireNonNull(definition, "definition must not be null");
+        Connection connection = connectionProvider.getConnection();
+        try {
+            if (!insertDefinitionRow(connection, definition)) {
+                updateDefinitionName(connection, definition);
+                deleteGraph(connection, definition.id());
+            }
+            insertGraph(connection, definition);
+        } catch (SQLException e) {
+            throw new JdbcStorageException("failed to upsert definition: " + definition.id(), e);
+        } finally {
+            connectionProvider.close(connection);
+        }
+    }
+
    @Override
    public Optional<ProcessDefinition> findById(String definitionId) {
        Connection connection = connectionProvider.getConnection();
@@ -157,6 +145,77 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
        }
    }

+    private static boolean insertDefinitionRow(Connection connection, ProcessDefinition definition) throws SQLException {
+        try (PreparedStatement insertDefinition = connection.prepareStatement(INSERT_DEFINITION)) {
+            insertDefinition.setString(1, definition.id());
+            insertDefinition.setString(2, definition.name());
+            insertDefinition.executeUpdate();
+            return true;
+        } catch (SQLException e) {
+            if (isDuplicateKey(e)) {
+                return false;
+            }
+            throw e;
+        }
+    }
+
+    private static void updateDefinitionName(Connection connection, ProcessDefinition definition) throws SQLException {
+        try (PreparedStatement update = connection.prepareStatement(UPDATE_DEFINITION_NAME)) {
+            update.setString(1, definition.name());
+            update.setString(2, definition.id());
+            update.executeUpdate();
+        }
+    }
+
+    private static void deleteGraph(Connection connection, String definitionId) throws SQLException {
+        try (PreparedStatement deleteTransitions = connection.prepareStatement(DELETE_TRANSITIONS)) {
+            deleteTransitions.setString(1, definitionId);
+            deleteTransitions.executeUpdate();
+        }
+        try (PreparedStatement deleteCandidates = connection.prepareStatement(DELETE_CANDIDATES)) {
+            deleteCandidates.setString(1, definitionId);
+            deleteCandidates.executeUpdate();
+        }
+        try (PreparedStatement deleteSteps = connection.prepareStatement(DELETE_STEPS)) {
+            deleteSteps.setString(1, definitionId);
+            deleteSteps.executeUpdate();
+        }
+    }
+
+    private static void insertGraph(Connection connection, ProcessDefinition definition) throws SQLException {
+        int stepOrder = 0;
+        for (ApprovalStep step : definition.steps()) {
+            try (PreparedStatement insertStep = connection.prepareStatement(INSERT_STEP)) {
+                insertStep.setString(1, definition.id());
+                insertStep.setString(2, step.id());
+                insertStep.setString(3, step.name());
+                insertStep.setString(4, step.policy().name());
+                insertStep.setInt(5, stepOrder++);
+                insertStep.executeUpdate();
+            }
+            int candidateOrder = 0;
+            for (String candidate : step.candidates()) {
+                try (PreparedStatement insertCandidate = connection.prepareStatement(INSERT_CANDIDATE)) {
+                    insertCandidate.setString(1, definition.id());
+                    insertCandidate.setString(2, step.id());
+                    insertCandidate.setString(3, candidate);
+                    insertCandidate.setInt(4, candidateOrder++);
+                    insertCandidate.executeUpdate();
+                }
+            }
+        }
+        for (StepTransition transition : definition.transitions()) {
+            try (PreparedStatement insertTransition = connection.prepareStatement(INSERT_TRANSITION)) {
+                insertTransition.setString(1, definition.id());
+                insertTransition.setString(2, transition.fromStepId());
+                insertTransition.setString(3, transition.toStepId());
+                insertTransition.setString(4, transition.conditionKey());
+                insertTransition.setInt(5, transition.priority());
+                insertTransition.executeUpdate();
+            }
+        }
+    }
+
    private static boolean isDuplicateKey(SQLException e) {
        return "23505".equals(e.getSQLState());
    }
@@ -1,6 +1,7 @@
 package com.jetlumen.ordo.storage.jdbc;

 import com.jetlumen.ordo.api.ProcessInstance;
+import com.jetlumen.ordo.api.ProcessStatus;
 import com.jetlumen.ordo.api.repository.ProcessInstanceRepository;
 import com.jetlumen.ordo.storage.jdbc.mapper.ProcessInstanceMapper;

@@ -21,6 +22,8 @@ public final class JdbcProcessInstanceRepository implements ProcessInstanceRepos
    private static final String SELECT_INSTANCE =
            "SELECT id, definition_id, initiator, status, context_json, started_at, finished_at"
                    + " FROM ordo_process_instance WHERE id = ?";
+    private static final String EXISTS_RUNNING =
+            "SELECT 1 FROM ordo_process_instance WHERE definition_id = ? AND status = ? LIMIT 1";

    private final JdbcConnectionProvider connectionProvider;

@@ -72,4 +75,20 @@ public final class JdbcProcessInstanceRepository implements ProcessInstanceRepos
            connectionProvider.close(connection);
        }
    }
+
+    @Override
+    public boolean existsRunning(String definitionId) {
+        Connection connection = connectionProvider.getConnection();
+        try (PreparedStatement select = connection.prepareStatement(EXISTS_RUNNING)) {
+            select.setString(1, definitionId);
+            select.setString(2, ProcessStatus.RUNNING.name());
+            try (ResultSet resultSet = select.executeQuery()) {
+                return resultSet.next();
+            }
+        } catch (SQLException e) {
+            throw new JdbcStorageException("failed to check running instances for definition: " + definitionId, e);
+        } finally {
+            connectionProvider.close(connection);
+        }
+    }
 }
@@ -11,6 +11,7 @@ import com.jetlumen.ordo.api.ProcessStatus;
 import com.jetlumen.ordo.api.RoutingCondition;
 import com.jetlumen.ordo.api.StepTransition;
 import com.jetlumen.ordo.api.TaskStatus;
+import com.jetlumen.ordo.api.exception.DefinitionInUseException;
 import com.jetlumen.ordo.api.exception.NoRouteFoundException;
 import com.jetlumen.ordo.api.exception.TaskAlreadyCompletedException;
 import com.jetlumen.ordo.core.DefaultOrdoEngine;
@@ -148,6 +149,22 @@ class JdbcOrdoEngineIntegrationTest {
        assertEquals("henry", engine.findPendingTasksByInstanceId(instance.id()).getFirst().assignee());
    }

+    @Test
+    void replaceSwapsTheGraphWhenNoInstanceIsRunning() {
+        engine.replace(ProcessDefinition.linear("leave", "Leave request v2", List.of(
+                ApprovalStep.single("director", "Director approval", "diana"))));
+
+        ProcessInstance instance = engine.start("leave", "alice");
+        assertEquals("diana", engine.findPendingTasksByInstanceId(instance.id()).getFirst().assignee());
+    }
+
+    @Test
+    void replaceIsRejectedWhileAnInstanceIsRunning() {
+        engine.start("leave", "alice");
+        assertThrows(DefinitionInUseException.class, () -> engine.replace(ProcessDefinition.linear(
+                "leave", "Leave request v2", List.of(ApprovalStep.single("director", "Director approval", "diana")))));
+    }
+
    private OrdoEngine newEngine(AssigneeResolver assigneeResolver) {
        return new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), assigneeResolver, RoutingCondition.always(),
                new JdbcTransactionExecutor(connectionProvider),
@@ -39,6 +39,20 @@ class JdbcProcessDefinitionRepositoryTest {
        assertEquals("Leave request v1", repository.findById("leave").orElseThrow().name());
    }

+    @Test
+    void upsertReplacesNameStepsCandidatesAndTransitions() {
+        repository.insertIfAbsent(ProcessDefinition.linear("leave", "Leave request v1", List.of(
+                ApprovalStep.single("manager", "Manager approval", "maria"),
+                ApprovalStep.single("hr", "HR approval", "henry"))));
+
+        ProcessDefinition replacement = new ProcessDefinition("leave", "Leave request v2", List.of(
+                ApprovalStep.single("director", "Director approval", "diana")),
+                List.of(StepTransition.end("director")));
+        repository.upsert(replacement);
+
+        assertEquals(replacement, repository.findById("leave").orElseThrow());
+    }
+
    @Test
    void returnsEmptyForAnUnknownDefinition() {
        assertTrue(repository.findById("missing").isEmpty());
@@ -13,6 +13,7 @@ import java.util.List;
 import java.util.Map;

 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;

@@ -66,6 +67,20 @@ class JdbcProcessInstanceRepositoryTest {
        assertTrue(repository.findById("missing").isEmpty());
    }

+    @Test
+    void existsRunningOnlyCountsRunningInstancesOfThatDefinition() {
+        assertFalse(repository.existsRunning("leave"));
+
+        repository.insert(new ProcessInstance("inst-1", "leave", "alice", ProcessStatus.RUNNING,
+                STARTED_AT, null, ProcessContext.empty()));
+        assertTrue(repository.existsRunning("leave"));
+        assertFalse(repository.existsRunning("other"));
+
+        repository.update(new ProcessInstance("inst-1", "leave", "alice", ProcessStatus.APPROVED,
+                STARTED_AT, STARTED_AT.plusSeconds(60), ProcessContext.empty()));
+        assertFalse(repository.existsRunning("leave"));
+    }
+
    @Test
    void updateOfAnUnknownInstanceFails() {
        assertThrows(IllegalStateException.class, () -> repository.update(new ProcessInstance("missing", "leave",