feat: Implement pagination and filtering for approval tasks and process instances

- Added query method to InMemoryApprovalTaskRepository for filtering and pagination of approval tasks. - Enhanced InMemoryProcessInstanceRepository with query method for filtering and pagination of process instances. - Introduced Page and PageRequest classes for handling paginated results. - Updated JdbcApprovalTaskRepository and JdbcProcessInstanceRepository to support querying with pagination. - Added tests for querying, filtering, and pagination in InMemory and JDBC repositories. - Implemented dynamic WHERE clause building for SQL queries in PageSupport.
2026-09-15 08:56:05 +08:00
parent e83234e98a
commit 77d1e5198b
26 changed files with 963 additions and 39 deletions
@@ -1,13 +1,18 @@
 package com.jetlumen.ordo.storage.jdbc;

 import com.jetlumen.ordo.api.ApprovalTask;
+import com.jetlumen.ordo.api.query.Page;
+import com.jetlumen.ordo.api.query.PageRequest;
+import com.jetlumen.ordo.api.query.TaskQuery;
 import com.jetlumen.ordo.api.repository.ApprovalTaskRepository;
+import com.jetlumen.ordo.storage.jdbc.PageSupport.WhereClause;
 import com.jetlumen.ordo.storage.jdbc.mapper.ApprovalTaskMapper;

 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -36,6 +41,9 @@ public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository
    private static final String SELECT_PENDING_BY_INSTANCE =
            "SELECT " + TASK_COLUMNS + " FROM ordo_approval_task WHERE status = 'PENDING' AND instance_id = ?"
                    + " ORDER BY created_at, id";
+    private static final String TASK_COLUMNS_QUALIFIED =
+            "t.id, t.instance_id, t.step_id, t.task_name, t.assignee, t.status, t.created_at, t.completed_at,"
+                    + " t.action_actor, t.action_comment, t.action_at";

    private final JdbcConnectionProvider connectionProvider;

@@ -111,6 +119,69 @@ public final class JdbcApprovalTaskRepository implements ApprovalTaskRepository
        }
    }

+    @Override
+    public Page<ApprovalTask> query(TaskQuery query, PageRequest pageRequest) {
+        Objects.requireNonNull(query, "query must not be null");
+        Objects.requireNonNull(pageRequest, "pageRequest must not be null");
+        WhereClause where = buildWhereClause(query);
+        Connection connection = connectionProvider.getConnection();
+        try {
+            long total = PageSupport.count(connection,
+                    "SELECT COUNT(*) FROM ordo_approval_task t" + where.sql(), where.params());
+            String sql = "SELECT " + TASK_COLUMNS_QUALIFIED + " FROM ordo_approval_task t" + where.sql()
+                    + " ORDER BY t.created_at DESC, t.id DESC LIMIT ? OFFSET ?";
+            List<ApprovalTask> content = new ArrayList<>();
+            try (PreparedStatement select = connection.prepareStatement(sql)) {
+                int index = PageSupport.bindParams(select, where.params());
+                select.setInt(index++, pageRequest.size());
+                select.setInt(index, pageRequest.offset());
+                try (ResultSet resultSet = select.executeQuery()) {
+                    while (resultSet.next()) {
+                        content.add(ApprovalTaskMapper.read(resultSet));
+                    }
+                }
+            }
+            return new Page<>(content, total, pageRequest.page(), pageRequest.size());
+        } catch (SQLException e) {
+            throw new JdbcStorageException("failed to query tasks", e);
+        } finally {
+            connectionProvider.close(connection);
+        }
+    }
+
+    private static WhereClause buildWhereClause(TaskQuery query) {
+        List<String> conditions = new ArrayList<>();
+        List<Object> params = new ArrayList<>();
+        if (query.assignee() != null) {
+            conditions.add("t.assignee = ?");
+            params.add(query.assignee());
+        }
+        if (query.instanceId() != null) {
+            conditions.add("t.instance_id = ?");
+            params.add(query.instanceId());
+        }
+        if (query.status() != null) {
+            conditions.add("t.status = ?");
+            params.add(query.status().name());
+        }
+        if (query.createdFrom() != null) {
+            conditions.add("t.created_at >= ?");
+            params.add(Timestamp.from(query.createdFrom()));
+        }
+        if (query.createdTo() != null) {
+            conditions.add("t.created_at <= ?");
+            params.add(Timestamp.from(query.createdTo()));
+        }
+        String join = "";
+        if (query.definitionId() != null) {
+            join = " JOIN ordo_process_instance i ON i.id = t.instance_id";
+            conditions.add("i.definition_id = ?");
+            params.add(query.definitionId());
+        }
+        String where = conditions.isEmpty() ? "" : " WHERE " + String.join(" AND ", conditions);
+        return new WhereClause(join + where, params);
+    }
+
    private Optional<ApprovalTask> findOne(String sql, String parameter) {
        Connection connection = connectionProvider.getConnection();
        try (PreparedStatement select = connection.prepareStatement(sql)) {
@@ -3,6 +3,8 @@ 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.query.Page;
+import com.jetlumen.ordo.api.query.PageRequest;
 import com.jetlumen.ordo.api.repository.ProcessDefinitionRepository;
 import com.jetlumen.ordo.storage.jdbc.mapper.ApprovalStepMapper;
 import com.jetlumen.ordo.storage.jdbc.mapper.ApprovalStepMapper.StepRow;
@@ -49,6 +51,8 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
    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 static final String SELECT_DEFINITIONS_PAGE =
+            "SELECT id, name FROM ordo_process_definition ORDER BY id LIMIT ? OFFSET ?";

    private final JdbcConnectionProvider connectionProvider;

@@ -112,43 +116,7 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
                    name = resultSet.getString("name");
                }
            }
-            List<StepRow> stepRows = new ArrayList<>();
-            try (PreparedStatement selectSteps = connection.prepareStatement(SELECT_STEPS)) {
-                selectSteps.setString(1, definitionId);
-                try (ResultSet resultSet = selectSteps.executeQuery()) {
-                    while (resultSet.next()) {
-                        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(), row.kind(),
-                        row.actionKey()));
-            }
-            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));
+            return Optional.of(assemble(connection, definitionId, name));
        } catch (SQLException e) {
            throw new JdbcStorageException("failed to load definition: " + definitionId, e);
        } finally {
@@ -156,6 +124,74 @@ public final class JdbcProcessDefinitionRepository implements ProcessDefinitionR
        }
    }

+    @Override
+    public Page<ProcessDefinition> findAll(PageRequest pageRequest) {
+        Objects.requireNonNull(pageRequest, "pageRequest must not be null");
+        Connection connection = connectionProvider.getConnection();
+        try {
+            long total = PageSupport.count(connection, "SELECT COUNT(*) FROM ordo_process_definition", List.of());
+            List<ProcessDefinition> content = new ArrayList<>();
+            try (PreparedStatement select = connection.prepareStatement(SELECT_DEFINITIONS_PAGE)) {
+                select.setInt(1, pageRequest.size());
+                select.setInt(2, pageRequest.offset());
+                List<Map.Entry<String, String>> idsAndNames = new ArrayList<>();
+                try (ResultSet resultSet = select.executeQuery()) {
+                    while (resultSet.next()) {
+                        idsAndNames.add(Map.entry(resultSet.getString("id"), resultSet.getString("name")));
+                    }
+                }
+                for (Map.Entry<String, String> idAndName : idsAndNames) {
+                    content.add(assemble(connection, idAndName.getKey(), idAndName.getValue()));
+                }
+            }
+            return new Page<>(content, total, pageRequest.page(), pageRequest.size());
+        } catch (SQLException e) {
+            throw new JdbcStorageException("failed to list definitions", e);
+        } finally {
+            connectionProvider.close(connection);
+        }
+    }
+
+    private static ProcessDefinition assemble(Connection connection, String definitionId, String name) throws SQLException {
+        List<StepRow> stepRows = new ArrayList<>();
+        try (PreparedStatement selectSteps = connection.prepareStatement(SELECT_STEPS)) {
+            selectSteps.setString(1, definitionId);
+            try (ResultSet resultSet = selectSteps.executeQuery()) {
+                while (resultSet.next()) {
+                    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(), row.kind(),
+                    row.actionKey()));
+        }
+        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 new ProcessDefinition(definitionId, name, steps, transitions);
+    }
+
    private static boolean definitionExists(Connection connection, String definitionId) throws SQLException {
        try (PreparedStatement select = connection.prepareStatement(SELECT_DEFINITION)) {
            select.setString(1, definitionId);
@@ -2,13 +2,20 @@ package com.jetlumen.ordo.storage.jdbc;

 import com.jetlumen.ordo.api.ProcessInstance;
 import com.jetlumen.ordo.api.ProcessStatus;
+import com.jetlumen.ordo.api.query.InstanceQuery;
+import com.jetlumen.ordo.api.query.Page;
+import com.jetlumen.ordo.api.query.PageRequest;
 import com.jetlumen.ordo.api.repository.ProcessInstanceRepository;
+import com.jetlumen.ordo.storage.jdbc.PageSupport.WhereClause;
 import com.jetlumen.ordo.storage.jdbc.mapper.ProcessInstanceMapper;

 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 import java.util.Optional;

@@ -107,4 +114,62 @@ public final class JdbcProcessInstanceRepository implements ProcessInstanceRepos
            connectionProvider.close(connection);
        }
    }
+
+    @Override
+    public Page<ProcessInstance> query(InstanceQuery query, PageRequest pageRequest) {
+        Objects.requireNonNull(query, "query must not be null");
+        Objects.requireNonNull(pageRequest, "pageRequest must not be null");
+        WhereClause where = buildWhereClause(query);
+        Connection connection = connectionProvider.getConnection();
+        try {
+            long total = PageSupport.count(connection,
+                    "SELECT COUNT(*) FROM ordo_process_instance" + where.sql(), where.params());
+            String sql = "SELECT id, definition_id, initiator, status, context_json, started_at, finished_at"
+                    + " FROM ordo_process_instance" + where.sql()
+                    + " ORDER BY started_at DESC, id DESC LIMIT ? OFFSET ?";
+            List<ProcessInstance> content = new ArrayList<>();
+            try (PreparedStatement select = connection.prepareStatement(sql)) {
+                int index = PageSupport.bindParams(select, where.params());
+                select.setInt(index++, pageRequest.size());
+                select.setInt(index, pageRequest.offset());
+                try (ResultSet resultSet = select.executeQuery()) {
+                    while (resultSet.next()) {
+                        content.add(ProcessInstanceMapper.read(resultSet));
+                    }
+                }
+            }
+            return new Page<>(content, total, pageRequest.page(), pageRequest.size());
+        } catch (SQLException e) {
+            throw new JdbcStorageException("failed to query instances", e);
+        } finally {
+            connectionProvider.close(connection);
+        }
+    }
+
+    private static WhereClause buildWhereClause(InstanceQuery query) {
+        List<String> conditions = new ArrayList<>();
+        List<Object> params = new ArrayList<>();
+        if (query.definitionId() != null) {
+            conditions.add("definition_id = ?");
+            params.add(query.definitionId());
+        }
+        if (query.status() != null) {
+            conditions.add("status = ?");
+            params.add(query.status().name());
+        }
+        if (query.initiator() != null) {
+            conditions.add("initiator = ?");
+            params.add(query.initiator());
+        }
+        if (query.startedFrom() != null) {
+            conditions.add("started_at >= ?");
+            params.add(Timestamp.from(query.startedFrom()));
+        }
+        if (query.startedTo() != null) {
+            conditions.add("started_at <= ?");
+            params.add(Timestamp.from(query.startedTo()));
+        }
+        String where = conditions.isEmpty() ? "" : " WHERE " + String.join(" AND ", conditions);
+        return new WhereClause(where, params);
+    }
 }
@@ -0,0 +1,35 @@
+package com.jetlumen.ordo.storage.jdbc;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+/** Shared helpers for building dynamic, paginated {@code WHERE} clauses across JDBC repositories. */
+final class PageSupport {
+    private PageSupport() {
+    }
+
+    /** A dynamically built {@code WHERE}/{@code JOIN} fragment (may be empty) plus its bind parameters, in order. */
+    record WhereClause(String sql, List<Object> params) {
+    }
+
+    static long count(Connection connection, String countSql, List<Object> params) throws SQLException {
+        try (PreparedStatement select = connection.prepareStatement(countSql)) {
+            bindParams(select, params);
+            try (ResultSet resultSet = select.executeQuery()) {
+                resultSet.next();
+                return resultSet.getLong(1);
+            }
+        }
+    }
+
+    static int bindParams(PreparedStatement statement, List<Object> params) throws SQLException {
+        int index = 1;
+        for (Object param : params) {
+            statement.setObject(index++, param);
+        }
+        return index;
+    }
+}