Elmma/ordo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add explicit conditional routing for approval steps

Browse Source 
Require ProcessDefinition transitions so the engine can branch or end by conditionKey, persist them in JDBC, and pin repository text files to LF.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
0264408
2026-09-10 18:56:59 +08:00
co-authored by Cursor
parent 581823290a
commit 58ec867eab
23 changed files with 594 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ordo-storage-jdbc/src/main/java/com/jetlumen/ordo/storage/jdbc/JdbcProcessDefinitionRepository.java
+30 -1
View File
@@ -2,9 +2,12 @@ package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.StepTransition;
import com.jetlumen.ordo.api.repository.ProcessDefinitionRepository;
import com.jetlumen.ordo.storage.jdbc.mapper.ApprovalStepMapper;
import com.jetlumen.ordo.storage.jdbc.mapper.ApprovalStepMapper.StepRow;
import com.jetlumen.ordo.storage.jdbc.mapper.StepTransitionMapper;
import com.jetlumen.ordo.storage.jdbc.mapper.StepTransitionMapper.TransitionRow;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -25,12 +28,17 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
"INSERT INTO ordo_approval_step (definition_id, step_id, step_name, policy, step_order) VALUES (?, ?, ?, ?, ?)";
private static final String INSERT_CANDIDATE =
"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 SELECT_DEFINITION =
"SELECT id, name FROM ordo_process_definition WHERE id = ?";
private static final String SELECT_STEPS =
"SELECT step_id, step_name, policy FROM ordo_approval_step WHERE definition_id = ? ORDER BY step_order";
private static final String SELECT_CANDIDATES =
"SELECT step_id, candidate FROM ordo_step_candidate WHERE definition_id = ? ORDER BY step_id, candidate_order";
private static final String SELECT_TRANSITIONS =
"SELECT from_step_id, to_step_id, condition_key, priority FROM ordo_step_transition "
+ "WHERE definition_id = ? ORDER BY from_step_id, priority";
private final JdbcConnectionProvider connectionProvider;
@@ -74,6 +82,16 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
}
}
}
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();
}
}
return true;
} catch (SQLException e) {
throw new JdbcStorageException("failed to insert definition: " + definition.id(), e);
@@ -120,7 +138,18 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
List<String> candidates = candidatesByStep.getOrDefault(row.stepId(), List.of());
steps.add(new ApprovalStep(row.stepId(), row.stepName(), candidates, row.policy()));
}
return Optional.of(new ProcessDefinition(definitionId, name, steps));
List<StepTransition> transitions = new ArrayList<>();
try (PreparedStatement selectTransitions = connection.prepareStatement(SELECT_TRANSITIONS)) {
selectTransitions.setString(1, definitionId);
try (ResultSet resultSet = selectTransitions.executeQuery()) {
while (resultSet.next()) {
TransitionRow row = StepTransitionMapper.readRow(resultSet);
transitions.add(new StepTransition(row.fromStepId(), row.toStepId(), row.conditionKey(),
row.priority()));
}
}
}
return Optional.of(new ProcessDefinition(definitionId, name, steps, transitions));
} catch (SQLException e) {
throw new JdbcStorageException("failed to load definition: " + definitionId, e);
} finally {
ordo-storage-jdbc/src/main/java/com/jetlumen/ordo/storage/jdbc/mapper/StepTransitionMapper.java
+21
View File
@@ -0,0 +1,21 @@
package com.jetlumen.ordo.storage.jdbc.mapper;
import java.sql.ResultSet;
import java.sql.SQLException;
/** Maps rows of {@code ordo_step_transition}. */
public final class StepTransitionMapper {
private StepTransitionMapper() {
}
public static TransitionRow readRow(ResultSet resultSet) throws SQLException {
return new TransitionRow(
resultSet.getString("from_step_id"),
resultSet.getString("to_step_id"),
resultSet.getString("condition_key"),
resultSet.getInt("priority"));
}
public record TransitionRow(String fromStepId, String toStepId, String conditionKey, int priority) {
}
}
ordo-storage-jdbc/src/main/resources/db/migration/V3__add_step_transitions.sql
+12
View File
@@ -0,0 +1,12 @@
CREATE TABLE ordo_step_transition (
definition_id VARCHAR(64) NOT NULL,
from_step_id VARCHAR(64) NOT NULL,
to_step_id VARCHAR(64),
condition_key VARCHAR(255),
priority INTEGER NOT NULL,
PRIMARY KEY (definition_id, from_step_id, priority),
CONSTRAINT fk_transition_from FOREIGN KEY (definition_id, from_step_id)
REFERENCES ordo_approval_step (definition_id, step_id),
CONSTRAINT fk_transition_to FOREIGN KEY (definition_id, to_step_id)
REFERENCES ordo_approval_step (definition_id, step_id)
);
ordo-storage-jdbc/src/test/java/com/jetlumen/ordo/storage/jdbc/JdbcApprovalTaskRepositoryTest.java
+1 -1
View File
@@ -38,7 +38,7 @@ class JdbcApprovalTaskRepositoryTest {
/** Tasks reference their instance, which references its definition; both parent rows must exist. */
private void insertFixtureData() {
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition("leave",
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear("leave",
"Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
JdbcProcessInstanceRepository instanceRepository = new JdbcProcessInstanceRepository(connectionProvider);
instanceRepository.insert(new ProcessInstance("inst-1", "leave", "alice", ProcessStatus.RUNNING,
ordo-storage-jdbc/src/test/java/com/jetlumen/ordo/storage/jdbc/JdbcOrdoEngineIntegrationTest.java
+28 -3
View File
@@ -8,7 +8,10 @@ import com.jetlumen.ordo.api.ProcessContext;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.ProcessInstance;
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.NoRouteFoundException;
import com.jetlumen.ordo.api.exception.TaskAlreadyCompletedException;
import com.jetlumen.ordo.core.DefaultOrdoEngine;
import org.junit.jupiter.api.BeforeEach;
@@ -35,7 +38,7 @@ class JdbcOrdoEngineIntegrationTest {
void setUp() {
connectionProvider = new JdbcConnectionProvider(JdbcTestSupport.newDataSource());
engine = newEngine(AssigneeResolver.direct());
engine.register(new ProcessDefinition("leave", "Leave request", List.of(
engine.register(ProcessDefinition.linear("leave", "Leave request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.single("hr", "HR approval", "henry"))));
}
@@ -94,7 +97,7 @@ class JdbcOrdoEngineIntegrationTest {
}
return candidate;
});
failingEngine.register(new ProcessDefinition("leave2", "Leave request", List.of(
failingEngine.register(ProcessDefinition.linear("leave2", "Leave request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.single("hr", "HR approval", "henry"))));
@@ -110,6 +113,28 @@ class JdbcOrdoEngineIntegrationTest {
assertEquals(ProcessStatus.RUNNING, failingEngine.findInstance(instance.id()).orElseThrow().status());
}
@Test
void rollsBackTheWholeApprovalWhenNoRouteMatches() {
OrdoEngine failingEngine = new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC),
AssigneeResolver.direct(), (key, context) -> false,
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
new JdbcApprovalTaskRepository(connectionProvider));
failingEngine.register(new ProcessDefinition("leave-noroute", "Leave request",
List.of(ApprovalStep.single("manager", "Manager approval", "maria")),
List.of(StepTransition.endWhen("manager", "never", 0))));
ProcessInstance instance = failingEngine.start("leave-noroute", "alice");
ApprovalTask managerTask = failingEngine.findPendingTasksByInstanceId(instance.id()).getFirst();
assertThrows(NoRouteFoundException.class, () -> failingEngine.approve(managerTask.id(), "maria"));
ApprovalTask storedTask = failingEngine.findTask(managerTask.id()).orElseThrow();
assertEquals(TaskStatus.PENDING, storedTask.status());
assertNull(storedTask.action());
assertEquals(ProcessStatus.RUNNING, failingEngine.findInstance(instance.id()).orElseThrow().status());
}
@Test
void dataSurvivesAcrossEngineInstancesOverTheSameDataSource() {
ProcessInstance instance = engine.start("leave", "alice");
@@ -124,7 +149,7 @@ class JdbcOrdoEngineIntegrationTest {
}
private OrdoEngine newEngine(AssigneeResolver assigneeResolver) {
return new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), assigneeResolver,
return new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), assigneeResolver, RoutingCondition.always(),
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
ordo-storage-jdbc/src/test/java/com/jetlumen/ordo/storage/jdbc/JdbcPostgresIntegrationTest.java
+9 -8
View File
@@ -8,6 +8,7 @@ import com.jetlumen.ordo.api.ProcessContext;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.ProcessInstance;
import com.jetlumen.ordo.api.ProcessStatus;
import com.jetlumen.ordo.api.RoutingCondition;
import com.jetlumen.ordo.api.TaskAction;
import com.jetlumen.ordo.api.TaskStatus;
import com.jetlumen.ordo.core.DefaultOrdoEngine;
@@ -102,7 +103,7 @@ class JdbcPostgresIntegrationTest {
@Test
void engineCompletesASequentialApprovalProcessOverPostgres() {
OrdoEngine engine = newEngine(AssigneeResolver.direct());
engine.register(new ProcessDefinition("leave-pg", "Leave request", List.of(
engine.register(ProcessDefinition.linear("leave-pg", "Leave request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.single("hr", "HR approval", "henry"))));
@@ -122,18 +123,18 @@ class JdbcPostgresIntegrationTest {
@Test
void rejectsDuplicateDefinitionIdsViaTheDatabaseUniqueConstraint() {
JdbcProcessDefinitionRepository repository = new JdbcProcessDefinitionRepository(connectionProvider);
ProcessDefinition definition = new ProcessDefinition("leave-dup-pg", "Leave request",
ProcessDefinition definition = ProcessDefinition.linear("leave-dup-pg", "Leave request",
List.of(ApprovalStep.single("manager", "Manager approval", "maria")));
assertTrue(repository.insertIfAbsent(definition));
assertFalse(repository.insertIfAbsent(new ProcessDefinition("leave-dup-pg", "Second attempt",
assertFalse(repository.insertIfAbsent(ProcessDefinition.linear("leave-dup-pg", "Second attempt",
List.of(ApprovalStep.single("manager", "Manager approval", "maria")))));
assertEquals("Leave request", repository.findById("leave-dup-pg").orElseThrow().name());
}
@Test
void rejectsDuplicateInstanceIdsViaTheDatabaseUniqueConstraint() {
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition(
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear(
"leave-dup-inst-pg", "Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
JdbcProcessInstanceRepository repository = new JdbcProcessInstanceRepository(connectionProvider);
ProcessInstance instance = new ProcessInstance("inst-dup-pg", "leave-dup-inst-pg", "alice",
@@ -145,7 +146,7 @@ class JdbcPostgresIntegrationTest {
@Test
void roundsTimestampsToMicrosecondPrecision() {
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition(
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear(
"leave-time-pg", "Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
JdbcProcessInstanceRepository repository = new JdbcProcessInstanceRepository(connectionProvider);
@@ -164,7 +165,7 @@ class JdbcPostgresIntegrationTest {
@Test
void onlyOneOfTwoConcurrentCompletionsWinsOnPostgres() throws Exception {
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition(
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear(
"leave-race-pg", "Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
JdbcProcessInstanceRepository instanceRepository = new JdbcProcessInstanceRepository(connectionProvider);
instanceRepository.insert(new ProcessInstance("inst-race-pg", "leave-race-pg", "alice",
@@ -200,7 +201,7 @@ class JdbcPostgresIntegrationTest {
}
return candidate;
});
failingEngine.register(new ProcessDefinition("leave-rollback-pg", "Leave request", List.of(
failingEngine.register(ProcessDefinition.linear("leave-rollback-pg", "Leave request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.single("hr", "HR approval", "henry"))));
@@ -217,7 +218,7 @@ class JdbcPostgresIntegrationTest {
}
private OrdoEngine newEngine(AssigneeResolver assigneeResolver) {
return new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), assigneeResolver,
return new DefaultOrdoEngine(Clock.fixed(NOW, ZoneOffset.UTC), assigneeResolver, RoutingCondition.always(),
new JdbcTransactionExecutor(connectionProvider),
new JdbcProcessDefinitionRepository(connectionProvider),
new JdbcProcessInstanceRepository(connectionProvider),
ordo-storage-jdbc/src/test/java/com/jetlumen/ordo/storage/jdbc/JdbcProcessDefinitionRepositoryTest.java
+17 -2
View File
@@ -2,6 +2,7 @@ package com.jetlumen.ordo.storage.jdbc;
import com.jetlumen.ordo.api.ApprovalStep;
import com.jetlumen.ordo.api.ProcessDefinition;
import com.jetlumen.ordo.api.StepTransition;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -22,7 +23,7 @@ class JdbcProcessDefinitionRepositoryTest {
@Test
void insertsAndReadsBackADefinitionWithItsStepsInOrder() {
ProcessDefinition definition = new ProcessDefinition("leave", "Leave request", List.of(
ProcessDefinition definition = ProcessDefinition.linear("leave", "Leave request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.single("hr", "HR approval", "henry")));
@@ -43,7 +44,21 @@ class JdbcProcessDefinitionRepositoryTest {
assertTrue(repository.findById("missing").isEmpty());
}
@Test
void insertsAndReadsBackConditionalTransitionsIncludingTerminalEdges() {
ProcessDefinition definition = new ProcessDefinition("expense", "Expense request", List.of(
ApprovalStep.single("manager", "Manager approval", "maria"),
ApprovalStep.single("director", "Director approval", "diana")),
List.of(
StepTransition.when("manager", "director", "amount-gt-1000", 0),
new StepTransition("manager", null, null, 1),
StepTransition.end("director")));
assertTrue(repository.insertIfAbsent(definition));
assertEquals(definition, repository.findById("expense").orElseThrow());
}
private static ProcessDefinition definition(String id, String name) {
return new ProcessDefinition(id, name, List.of(ApprovalStep.single("lead", "Lead approval", "lee")));
return ProcessDefinition.linear(id, name, List.of(ApprovalStep.single("lead", "Lead approval", "lee")));
}
}
ordo-storage-jdbc/src/test/java/com/jetlumen/ordo/storage/jdbc/JdbcProcessInstanceRepositoryTest.java
+1 -1
View File
@@ -27,7 +27,7 @@ class JdbcProcessInstanceRepositoryTest {
connectionProvider = new JdbcConnectionProvider(JdbcTestSupport.newDataSource());
repository = new JdbcProcessInstanceRepository(connectionProvider);
// instances reference their definition, so the parent row must exist
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition("leave",
new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(ProcessDefinition.linear("leave",
"Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
}
ordo-storage-jdbc/src/test/java/com/jetlumen/ordo/storage/jdbc/JdbcTestSupport.java
+2 -1
View File
@@ -15,7 +15,8 @@ import java.util.UUID;
final class JdbcTestSupport {
private static final String[] MIGRATIONS = {
"/db/migration/V1__create_ordo_tables.sql",
"/db/migration/V2__add_step_candidates_and_policy.sql"
"/db/migration/V2__add_step_candidates_and_policy.sql",
"/db/migration/V3__add_step_transitions.sql"
};
private static final String[] SCHEMA_SQL = loadSchemas();