feat: support multi-approver steps with ANY/ALL policy

- ApprovalStep now has candidates + ApprovalPolicy (ANY/ALL) instead of a single assignee; ApprovalStep.single(...) kept as a convenience factory for the single-approver case - AssigneeResolver resolves per-candidate; TaskStatus gains SKIPPED for auto-skipped sibling tasks - DefaultOrdoEngine creates one task per candidate and advances per-step according to policy (ANY: first approval wins and skips the rest, only rejects once everyone rejects; ALL: fail-fast on first rejection, only advances once everyone approves) - Added ApprovalTaskRepository.findByInstanceIdAndStepId and JDBC/in-memory implementations - Added V2 migration: ordo_approval_step.policy column + ordo_step_candidate table - Updated ApprovalStepMapper/ApprovalTaskMapper/JdbcProcessDefinitionRepository for the new schema, handling null action for SKIPPED tasks - Migrated all call sites (tests, ordo-example, rhizome demo) to ApprovalStep.single(...) and the 3-arg AssigneeResolver - Added ANY/ALL policy test coverage; fixed JdbcTestSupport to apply the V2 migration and strip SQL comments before splitting on ';'
2026-09-10 18:20:33 +08:00
parent c524c74289
commit 581823290a
22 changed files with 440 additions and 90 deletions
@@ -27,6 +27,9 @@ public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository
            "SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE id = ?";
    private static final String SELECT_BY_INSTANCE =
            "SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE instance_id = ? ORDER BY created_at, id";
+    private static final String SELECT_BY_INSTANCE_AND_STEP =
+            "SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE instance_id = ? AND step_id = ?"
+                    + " ORDER BY created_at, id";
    private static final String SELECT_PENDING_BY_ASSIGNEE =
            "SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE status = 'PENDING' AND assignee = ?"
                    + " ORDER BY created_at, id";
@@ -64,6 +67,26 @@ public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository
        return findAll(SELECT_BY_INSTANCE, instanceId);
    }

+    @Override
+    public List<ApprovalTask> findByInstanceIdAndStepId(String instanceId, String stepId) {
+        Connection connection = connectionProvider.getConnection();
+        try (PreparedStatement select = connection.prepareStatement(SELECT_BY_INSTANCE_AND_STEP)) {
+            select.setString(1, instanceId);
+            select.setString(2, stepId);
+            try (ResultSet resultSet = select.executeQuery()) {
+                List<ApprovalTask> tasks = new ArrayList<>();
+                while (resultSet.next()) {
+                    tasks.add(ApprovalTaskMapper.read(resultSet));
+                }
+                return tasks;
+            }
+        } catch (SQLException e) {
+            throw new JdbcStorageException("failed to query tasks for step: " + stepId, e);
+        } finally {
+            connectionProvider.close(connection);
+        }
+    }
+
    @Override
    public List<ApprovalTask> findPendingByAssignee(String assignee) {
        return findAll(SELECT_PENDING_BY_ASSIGNEE, assignee);
@@ -4,26 +4,33 @@ import com.jetlumen.ordo.api.ApprovalStep;
 import com.jetlumen.ordo.api.ProcessDefinition;
 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 java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;

-/** JDBC implementation of the definition storage port; steps live in a separate table. */
+/** JDBC implementation of the definition storage port; steps and their candidates live in separate tables. */
 public final class JdbcProcessDefinitionRepository implements ProcessDefinitionRepository {
    private static final String INSERT_DEFINITION =
            "INSERT INTO ordo_process_definition (id, name) VALUES (?, ?)";
    private static final String INSERT_STEP =
-            "INSERT INTO ordo_approval_step (definition_id, step_id, step_name, assignee, step_order) VALUES (?, ?, ?, ?, ?)";
+            "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 SELECT_DEFINITION =
            "SELECT id, name FROM ordo_process_definition WHERE id = ?";
    private static final String SELECT_STEPS =
-            "SELECT step_id, step_name, assignee FROM ordo_approval_step WHERE definition_id = ? ORDER BY step_order";
+            "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 final JdbcConnectionProvider connectionProvider;

@@ -46,16 +53,26 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
                }
                throw new JdbcStorageException("failed to insert definition: " + definition.id(), e);
            }
-            int order = 0;
+            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.assignee());
-                    insertStep.setInt(5, order++);
+                    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();
+                    }
+                }
            }
            return true;
        } catch (SQLException e) {
@@ -79,15 +96,30 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
                    name = resultSet.getString("name");
                }
            }
-            List<ApprovalStep> steps = new ArrayList<>();
+            List<StepRow> stepRows = new ArrayList<>();
            try (PreparedStatement selectSteps = connection.prepareStatement(SELECT_STEPS)) {
                selectSteps.setString(1, definitionId);
                try (ResultSet resultSet = selectSteps.executeQuery()) {
                    while (resultSet.next()) {
-                        steps.add(ApprovalStepMapper.read(resultSet));
+                        stepRows.add(ApprovalStepMapper.readRow(resultSet));
                    }
                }
            }
+            Map<String, List<String>> candidatesByStep = new LinkedHashMap<>();
+            try (PreparedStatement selectCandidates = connection.prepareStatement(SELECT_CANDIDATES)) {
+                selectCandidates.setString(1, definitionId);
+                try (ResultSet resultSet = selectCandidates.executeQuery()) {
+                    while (resultSet.next()) {
+                        candidatesByStep.computeIfAbsent(resultSet.getString("step_id"), key -> new ArrayList<>())
+                                .add(resultSet.getString("candidate"));
+                    }
+                }
+            }
+            List<ApprovalStep> steps = new ArrayList<>();
+            for (StepRow row : stepRows) {
+                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));
        } catch (SQLException e) {
            throw new JdbcStorageException("failed to load definition: " + definitionId, e);
@@ -1,17 +1,24 @@
 package com.jetlumen.ordo.storage.jdbc.mapper;

-import com.jetlumen.ordo.api.ApprovalStep;
+import com.jetlumen.ordo.api.ApprovalPolicy;

 import java.sql.ResultSet;
 import java.sql.SQLException;

-/** Maps rows of {@code ordo_approval_step} to {@link ApprovalStep} objects. */
+/**
+ * Maps rows of {@code ordo_approval_step}. A step's candidate list lives in the separate
+ * {@code ordo_step_candidate} table, so this only reads the scalar step columns; the repository
+ * assembles the full {@link com.jetlumen.ordo.api.ApprovalStep} once candidates are loaded.
+ */
 public final class ApprovalStepMapper {
    private ApprovalStepMapper() {
    }

-    public static ApprovalStep read(ResultSet resultSet) throws SQLException {
-        return new ApprovalStep(resultSet.getString("step_id"), resultSet.getString("step_name"),
-                resultSet.getString("assignee"));
+    public static StepRow readRow(ResultSet resultSet) throws SQLException {
+        return new StepRow(resultSet.getString("step_id"), resultSet.getString("step_name"),
+                ApprovalPolicy.valueOf(resultSet.getString("policy")));
+    }
+
+    public record StepRow(String stepId, String stepName, ApprovalPolicy policy) {
    }
 }
@@ -8,6 +8,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Timestamp;
+import java.sql.Types;

 /** Maps rows of {@code ordo_approval_task} to {@link ApprovalTask} objects and back. */
 public final class ApprovalTaskMapper {
@@ -28,9 +29,16 @@ public final class ApprovalTaskMapper {
        TaskAction action = completedTask.action();
        statement.setString(1, completedTask.status().name());
        statement.setTimestamp(2, Timestamp.from(completedTask.completedAt()));
-        statement.setString(3, action.actor());
-        statement.setString(4, action.comment());
-        statement.setTimestamp(5, Timestamp.from(action.operatedAt()));
+        if (action == null) {
+            // SKIPPED tasks have no actor - a sibling candidate's decision already settled the step.
+            statement.setNull(3, Types.VARCHAR);
+            statement.setNull(4, Types.VARCHAR);
+            statement.setNull(5, Types.TIMESTAMP);
+        } else {
+            statement.setString(3, action.actor());
+            statement.setString(4, action.comment());
+            statement.setTimestamp(5, Timestamp.from(action.operatedAt()));
+        }
        statement.setString(6, completedTask.id());
    }

@@ -0,0 +1,23 @@
+-- Adds multi-candidate (any/all) approval step support.
+-- A step no longer has a single fixed assignee; instead it has one or more
+-- candidates in ordo_step_candidate, and a policy column decides whether any
+-- single candidate approval is enough (ANY) or every candidate must approve (ALL).
+
+ALTER TABLE ordo_approval_step ADD COLUMN policy VARCHAR(16) NOT NULL DEFAULT 'ANY';
+
+CREATE TABLE ordo_step_candidate (
+    definition_id   VARCHAR(64)  NOT NULL,
+    step_id         VARCHAR(64)  NOT NULL,
+    candidate       VARCHAR(255) NOT NULL,
+    candidate_order INTEGER      NOT NULL,
+    PRIMARY KEY (definition_id, step_id, candidate),
+    CONSTRAINT fk_step_candidate_step FOREIGN KEY (definition_id, step_id)
+        REFERENCES ordo_approval_step (definition_id, step_id)
+);
+
+-- Migrate any existing single-assignee steps into the new candidate table before
+-- the now-unused column is dropped.
+INSERT INTO ordo_step_candidate (definition_id, step_id, candidate, candidate_order)
+SELECT definition_id, step_id, assignee, 0 FROM ordo_approval_step;
+
+ALTER TABLE ordo_approval_step DROP COLUMN assignee;
@@ -39,7 +39,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",
-                "Leave request", List.of(new ApprovalStep("manager", "Manager approval", "maria"))));
+                "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,
                CREATED_AT, null, ProcessContext.empty()));
@@ -36,8 +36,8 @@ class JdbcOrdoEngineIntegrationTest {
        connectionProvider = new JdbcConnectionProvider(JdbcTestSupport.newDataSource());
        engine = newEngine(AssigneeResolver.direct());
        engine.register(new ProcessDefinition("leave", "Leave request", List.of(
-                new ApprovalStep("manager", "Manager approval", "maria"),
-                new ApprovalStep("hr", "HR approval", "henry"))));
+                ApprovalStep.single("manager", "Manager approval", "maria"),
+                ApprovalStep.single("hr", "HR approval", "henry"))));
    }

    @Test
@@ -88,15 +88,15 @@ class JdbcOrdoEngineIntegrationTest {

    @Test
    void rollsBackTheWholeApprovalWhenTheNextStepCannotBeCreated() {
-        OrdoEngine failingEngine = newEngine((step, context) -> {
+        OrdoEngine failingEngine = newEngine((candidate, step, context) -> {
            if (step.id().equals("hr")) {
                throw new IllegalStateException("no hr approval today");
            }
-            return step.assignee();
+            return candidate;
        });
        failingEngine.register(new ProcessDefinition("leave2", "Leave request", List.of(
-                new ApprovalStep("manager", "Manager approval", "maria"),
-                new ApprovalStep("hr", "HR approval", "henry"))));
+                ApprovalStep.single("manager", "Manager approval", "maria"),
+                ApprovalStep.single("hr", "HR approval", "henry"))));

        ProcessInstance instance = failingEngine.start("leave2", "alice");
        ApprovalTask managerTask = failingEngine.findPendingTasksByInstanceId(instance.id()).getFirst();
@@ -103,8 +103,8 @@ class JdbcPostgresIntegrationTest {
    void engineCompletesASequentialApprovalProcessOverPostgres() {
        OrdoEngine engine = newEngine(AssigneeResolver.direct());
        engine.register(new ProcessDefinition("leave-pg", "Leave request", List.of(
-                new ApprovalStep("manager", "Manager approval", "maria"),
-                new ApprovalStep("hr", "HR approval", "henry"))));
+                ApprovalStep.single("manager", "Manager approval", "maria"),
+                ApprovalStep.single("hr", "HR approval", "henry"))));

        ProcessInstance instance = engine.start("leave-pg", "alice",
                new ProcessContext(Map.of("requestId", "LEAVE-2026-001", "days", 5)));
@@ -123,18 +123,18 @@ class JdbcPostgresIntegrationTest {
    void rejectsDuplicateDefinitionIdsViaTheDatabaseUniqueConstraint() {
        JdbcProcessDefinitionRepository repository = new JdbcProcessDefinitionRepository(connectionProvider);
        ProcessDefinition definition = new ProcessDefinition("leave-dup-pg", "Leave request",
-                List.of(new ApprovalStep("manager", "Manager approval", "maria")));
+                List.of(ApprovalStep.single("manager", "Manager approval", "maria")));

        assertTrue(repository.insertIfAbsent(definition));
        assertFalse(repository.insertIfAbsent(new ProcessDefinition("leave-dup-pg", "Second attempt",
-                List.of(new ApprovalStep("manager", "Manager approval", "maria")))));
+                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(
-                "leave-dup-inst-pg", "Leave request", List.of(new ApprovalStep("manager", "Manager approval", "maria"))));
+                "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",
                ProcessStatus.RUNNING, NOW, null, ProcessContext.empty());
@@ -146,7 +146,7 @@ class JdbcPostgresIntegrationTest {
    @Test
    void roundsTimestampsToMicrosecondPrecision() {
        new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition(
-                "leave-time-pg", "Leave request", List.of(new ApprovalStep("manager", "Manager approval", "maria"))));
+                "leave-time-pg", "Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
        JdbcProcessInstanceRepository repository = new JdbcProcessInstanceRepository(connectionProvider);

        Instant microAligned = Instant.parse("2026-01-15T09:00:00.123456Z");
@@ -165,7 +165,7 @@ class JdbcPostgresIntegrationTest {
    @Test
    void onlyOneOfTwoConcurrentCompletionsWinsOnPostgres() throws Exception {
        new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition(
-                "leave-race-pg", "Leave request", List.of(new ApprovalStep("manager", "Manager approval", "maria"))));
+                "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",
                ProcessStatus.RUNNING, NOW, null, ProcessContext.empty()));
@@ -194,15 +194,15 @@ class JdbcPostgresIntegrationTest {

    @Test
    void rollsBackTheWholeApprovalWhenTheNextStepCannotBeCreated() {
-        OrdoEngine failingEngine = newEngine((step, context) -> {
+        OrdoEngine failingEngine = newEngine((candidate, step, context) -> {
            if (step.id().equals("hr")) {
                throw new IllegalStateException("no hr approval today");
            }
-            return step.assignee();
+            return candidate;
        });
        failingEngine.register(new ProcessDefinition("leave-rollback-pg", "Leave request", List.of(
-                new ApprovalStep("manager", "Manager approval", "maria"),
-                new ApprovalStep("hr", "HR approval", "henry"))));
+                ApprovalStep.single("manager", "Manager approval", "maria"),
+                ApprovalStep.single("hr", "HR approval", "henry"))));

        ProcessInstance instance = failingEngine.start("leave-rollback-pg", "alice");
        ApprovalTask managerTask = failingEngine.findPendingTasksByInstanceId(instance.id()).getFirst();
@@ -23,8 +23,8 @@ class JdbcProcessDefinitionRepositoryTest {
    @Test
    void insertsAndReadsBackADefinitionWithItsStepsInOrder() {
        ProcessDefinition definition = new ProcessDefinition("leave", "Leave request", List.of(
-                new ApprovalStep("manager", "Manager approval", "maria"),
-                new ApprovalStep("hr", "HR approval", "henry")));
+                ApprovalStep.single("manager", "Manager approval", "maria"),
+                ApprovalStep.single("hr", "HR approval", "henry")));

        assertTrue(repository.insertIfAbsent(definition));
        assertEquals(definition, repository.findById("leave").orElseThrow());
@@ -44,6 +44,6 @@ class JdbcProcessDefinitionRepositoryTest {
    }

    private static ProcessDefinition definition(String id, String name) {
-        return new ProcessDefinition(id, name, List.of(new ApprovalStep("lead", "Lead approval", "lee")));
+        return new ProcessDefinition(id, name, List.of(ApprovalStep.single("lead", "Lead approval", "lee")));
    }
 }
@@ -28,7 +28,7 @@ class JdbcProcessInstanceRepositoryTest {
        repository = new JdbcProcessInstanceRepository(connectionProvider);
        // instances reference their definition, so the parent row must exist
        new JdbcProcessDefinitionRepository(connectionProvider).insertIfAbsent(new ProcessDefinition("leave",
-                "Leave request", List.of(new ApprovalStep("manager", "Manager approval", "maria"))));
+                "Leave request", List.of(ApprovalStep.single("manager", "Manager approval", "maria"))));
    }

    @Test
@@ -13,7 +13,11 @@ import java.util.UUID;

 /** Creates isolated in-memory H2 databases with the Ordo schema applied. */
 final class JdbcTestSupport {
-    private static final String SCHEMA_SQL = loadSchema();
+    private static final String[] MIGRATIONS = {
+            "/db/migration/V1__create_ordo_tables.sql",
+            "/db/migration/V2__add_step_candidates_and_policy.sql"
+    };
+    private static final String[] SCHEMA_SQL = loadSchemas();

    private JdbcTestSupport() {
    }
@@ -27,12 +31,14 @@ final class JdbcTestSupport {
        return dataSource;
    }

-    /** Applies the V1 migration script to an empty database, e.g. a PostgreSQL test container. */
+    /** Applies every migration script in order to an empty database, e.g. a PostgreSQL test container. */
    static void applySchema(DataSource dataSource) {
        try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement()) {
-            for (String sql : SCHEMA_SQL.split(";")) {
-                if (!sql.isBlank()) {
-                    statement.execute(sql);
+            for (String migration : SCHEMA_SQL) {
+                for (String sql : stripComments(migration).split(";")) {
+                    if (!sql.isBlank()) {
+                        statement.execute(sql);
+                    }
                }
            }
        } catch (SQLException e) {
@@ -40,14 +46,36 @@ final class JdbcTestSupport {
        }
    }

-    private static String loadSchema() {
-        try (InputStream input = JdbcTestSupport.class.getResourceAsStream("/db/migration/V1__create_ordo_tables.sql")) {
+    /**
+     * Removes {@code --} line comments before splitting on {@code ;}, since a semicolon inside a
+     * comment (e.g. "assignee; instead of...") would otherwise be mistaken for a statement break.
+     */
+    private static String stripComments(String sql) {
+        StringBuilder result = new StringBuilder();
+        for (String line : sql.split("\\r?\\n")) {
+            int commentIndex = line.indexOf("--");
+            result.append(commentIndex >= 0 ? line.substring(0, commentIndex) : line).append('\n');
+        }
+        return result.toString();
+    }
+
+    private static String[] loadSchemas() {
+        String[] schemas = new String[MIGRATIONS.length];
+        for (int i = 0; i < MIGRATIONS.length; i++) {
+            schemas[i] = loadSchema(MIGRATIONS[i]);
+        }
+        return schemas;
+    }
+
+    private static String loadSchema(String resourcePath) {
+        try (InputStream input = JdbcTestSupport.class.getResourceAsStream(resourcePath)) {
            if (input == null) {
-                throw new IllegalStateException("V1__create_ordo_tables.sql not found on the classpath");
+                throw new IllegalStateException(resourcePath + " not found on the classpath");
            }
            return new String(input.readAllBytes(), StandardCharsets.UTF_8);
        } catch (IOException e) {
-            throw new IllegalStateException("failed to read V1__create_ordo_tables.sql", e);
+            throw new IllegalStateException("failed to read " + resourcePath, e);
        }
    }
 }
+