Class JdbcFlowRepository

java.lang.Object
cloud.opencode.base.flow.persistence.jdbc.JdbcFlowRepository
All Implemented Interfaces:
FlowRepository
public final class JdbcFlowRepository extends Object implements FlowRepository
JDBC-backed FlowRepository for MySQL, PostgreSQL and H2. 基于 JDBC 的 FlowRepository 实现，支持 MySQL / PostgreSQL / H2。

SQL-injection posture (design v4 §6, §14):

  • The only non-parameterised string permitted in any SQL is the table name; that string is composed exclusively from the opencode.base.flow.table-prefix system property (defaults to "flow_"), validated against SqlBuilder.PREFIX_PATTERN at startup. Any mismatch throws FlowException with FlowErrorCode.CFG_INVALID_VALUE.
  • Every dynamic value flows through a PreparedStatement ? marker.
  • initSchema() executes one CREATE TABLE IF NOT EXISTS per statement (re-runnable); the script is shipped under schema/V1__init_<dialect>.sql in the JAR.

SQL 注入防护（设计 v4 §6 / §14）：

  • SQL 中唯一允许的非参数化字符串是表名；该字符串仅由系统属性 opencode.base.flow.table-prefix（默认 "flow_"） 构成，启动时按 SqlBuilder.PREFIX_PATTERN 严校验，不匹配抛 FlowException + FlowErrorCode.CFG_INVALID_VALUE。
  • 其它所有动态值均经 PreparedStatement ? 占位符传入。
  • initSchema() 逐句执行 CREATE TABLE IF NOT EXISTS （可重复执行）；脚本随 JAR 打包在 schema/V1__init_<dialect>.sql。
Since:
JDK 25, opencode-base-flow V1.0.5
Author:
Leon Soo

  • Field Details

    • SYS_PROP_TABLE_PREFIX

      public static final String SYS_PROP_TABLE_PREFIX
      System property used to override the table prefix at startup (default "flow_"). The value is validated against SqlBuilder.PREFIX_PATTERN. 启动时覆盖表前缀的系统属性（默认 "flow_"）。
      See Also:

    • DEFAULT_TABLE_PREFIX

      public static final String DEFAULT_TABLE_PREFIX
      Default table prefix when SYS_PROP_TABLE_PREFIX is unset. 当 SYS_PROP_TABLE_PREFIX 未设置时使用的默认前缀。
      See Also:

  • Constructor Details

    • JdbcFlowRepository

      public JdbcFlowRepository(DataSource dataSource)
      Constructs a repository backed by the supplied DataSource; the dialect is auto-detected from the connection's JDBC URL and the table prefix is read from the SYS_PROP_TABLE_PREFIX system property. 使用给定的 DataSource 构造仓储；方言由 JDBC URL 自动推断， 表前缀读自系统属性 SYS_PROP_TABLE_PREFIX。
      Parameters:
      dataSource - the data source — 数据源
      Throws:
      FlowException - with FlowErrorCode.CFG_INVALID_VALUE when the prefix is illegal or the JDBC URL cannot be probed — 前缀非法或 JDBC URL 无法探测时抛
      Since:
      JDK 25, opencode-base-flow V1.0.5

    • JdbcFlowRepository

      public JdbcFlowRepository(DataSource dataSource, String tablePrefix)
      Test-friendly constructor that takes the table prefix explicitly. 测试友好构造器，显式传入表前缀。
      Parameters:
      dataSource - the data source — 数据源
      tablePrefix - table prefix (e.g. "flow_") — 表前缀
      Throws:
      FlowException - with FlowErrorCode.CFG_INVALID_VALUE when the prefix is illegal or the JDBC URL cannot be probed
      Since:
      JDK 25, opencode-base-flow V1.0.5

  • Method Details

    • initSchema

      public void initSchema()
      Loads schema/V1__init_<dialect>.sql from the classpath, substitutes the validated ${TABLE_PREFIX} token, then executes each statement inside a single transaction (R1 N-4). Idempotent — uses CREATE TABLE IF NOT EXISTS. 加载 classpath 中的 schema/V1__init_<dialect>.sql，将 ${TABLE_PREFIX} 占位符替换为校验后的前缀，在一个事务内按顺序执行每条 SQL（R1 N-4）。 幂等 —— 使用 CREATE TABLE IF NOT EXISTS。

      MySQL caveat: MySQL implicitly commits each DDL statement (CREATE TABLE / CREATE INDEX), so rollback cannot undo partially executed scripts. Operators must clean up half-applied schemas manually when initSchema fails mid-way on MySQL. H2 and PostgreSQL respect the transaction boundary as expected.

      MySQL 注意：MySQL 的 DDL 隐式提交，事务回滚无法 撤销半执行的脚本；运维需手动清理。H2 / PostgreSQL 正常遵守事务边界。

      Throws:
      FlowException - when the script cannot be loaded or executed — 脚本加载或执行失败时抛
      Since:
      JDK 25, opencode-base-flow V1.0.5

    • dialect

      public JdbcDialect dialect()
      Returns the detected dialect (visible for diagnostics and testing). 返回探测出的方言（用于诊断与测试）。
      Returns:
      the dialect — 方言
      Since:
      JDK 25, opencode-base-flow V1.0.5

    • tablePrefix

      public String tablePrefix()
      Returns the validated table prefix (visible for diagnostics and testing). 返回校验后的表前缀（用于诊断与测试）。
      Returns:
      the prefix — 前缀
      Since:
      JDK 25, opencode-base-flow V1.0.5

    • saveDefinition

      public FlowDefinition saveDefinition(FlowDefinition def)
      Description copied from interface: FlowRepository
      Persists a new flow definition. The returned record reflects the stored state (including any engine-assigned fields). 持久化一份新的流程定义。返回已存储的视图。
      Specified by:
      saveDefinition in interface FlowRepository
      Parameters:
      def - the definition to persist — 待持久化的定义
      Returns:
      the persisted definition — 已持久化的定义

    • findDefinitionById

      public Optional<FlowDefinition> findDefinitionById(String id)
      Description copied from interface: FlowRepository
      Looks up a definition by its primary id. 根据主键 ID 查找定义。
      Specified by:
      findDefinitionById in interface FlowRepository
      Parameters:
      id - definition id — 定义 ID
      Returns:
      the definition wrapped in an Optional — 定义（Optional）

    • findActiveDefinition

      public Optional<FlowDefinition> findActiveDefinition(String flowKey)
      Description copied from interface: FlowRepository
      Returns the single FlowStatus.ACTIVE definition for the given business key, if any. 返回指定业务 key 当前唯一启用的（FlowStatus.ACTIVE）定义。
      Specified by:
      findActiveDefinition in interface FlowRepository
      Parameters:
      flowKey - business flow key — 业务 key
      Returns:
      the active definition — 启用中的定义

    • findDefinitionVersions

      public List<FlowDefinition> findDefinitionVersions(String flowKey)
      Description copied from interface: FlowRepository
      Lists every persisted version of the given business key, in version ascending order. 返回给定业务 key 所有版本（按版本号升序）。
      Specified by:
      findDefinitionVersions in interface FlowRepository
      Parameters:
      flowKey - business flow key — 业务 key
      Returns:
      all definition versions — 全部版本

    • updateDefinitionStatus

      public void updateDefinitionStatus(String id, FlowStatus newStatus)
      Description copied from interface: FlowRepository
      Mutates the lifecycle status of a definition (ACTIVE / DEPRECATED). 修改某定义的生命周期状态（ACTIVE / DEPRECATED）。
      Specified by:
      updateDefinitionStatus in interface FlowRepository
      Parameters:
      id - definition id — 定义 ID
      newStatus - the new status — 新状态

    • saveInstance

      public FlowInstance saveInstance(FlowInstance inst)
      Description copied from interface: FlowRepository
      Persists a new flow instance. 持久化新的流程实例。
      Specified by:
      saveInstance in interface FlowRepository
      Parameters:
      inst - the instance to persist — 待持久化的实例
      Returns:
      the persisted instance — 已持久化的实例

    • findInstanceById

      public Optional<FlowInstance> findInstanceById(String id)
      Description copied from interface: FlowRepository
      Looks up an instance by id. 根据 ID 查找实例。
      Specified by:
      findInstanceById in interface FlowRepository
      Parameters:
      id - instance id — 实例 ID
      Returns:
      the instance wrapped in an Optional — 实例（Optional）

    • findInstancesByInitiator

      public List<FlowInstance> findInstancesByInitiator(String userId, int page, int size)
      Description copied from interface: FlowRepository
      Lists instances started by the given user, paginated. 分页列出指定发起人启动的实例。
      Specified by:
      findInstancesByInitiator in interface FlowRepository
      Parameters:
      userId - initiator user id — 发起人 userId
      page - zero-based page index — 页码（0 起）
      size - page size — 每页大小
      Returns:
      instances on the requested page — 该页实例

    • updateInstance

      public int updateInstance(FlowInstance inst)
      Description copied from interface: FlowRepository
      Optimistic-lock UPDATE; returns the number of affected rows (0 indicates version conflict). 乐观锁 UPDATE；返回 affected rows（0 表示版本冲突）。
      Specified by:
      updateInstance in interface FlowRepository
      Parameters:
      inst - the instance to update — 待更新的实例
      Returns:
      affected rows — 影响行数

    • casIncrementAddsignDepth

      public int casIncrementAddsignDepth(String instanceId, int expectedVersion, int maxDepth)
      Description copied from interface: FlowRepository
      Compare-and-set: UPDATE addsign_depth = addsign_depth + 1, version = version + 1 WHERE id = ? AND version = ? AND addsign_depth &lt; maxDepth. Returns affected rows (0 = conflict or depth exceeded). Callers must re-read the instance to distinguish the two cases. CAS：UPDATE addsign_depth + 1, version + 1 WHERE id = ? AND version = ? AND addsign_depth &lt; maxDepth。返回 affected rows（0 表示冲突或超深）。调用方需复读 instance 区分两种情况。
      Specified by:
      casIncrementAddsignDepth in interface FlowRepository
      Parameters:
      instanceId - instance id — 实例 ID
      expectedVersion - expected current version — 期望的当前版本
      maxDepth - configured max addsign depth — 配置的加签链上限
      Returns:
      affected rows — 影响行数

    • saveTask

      public Task saveTask(Task task)
      Description copied from interface: FlowRepository
      Persists a new task. 持久化新任务。
      Specified by:
      saveTask in interface FlowRepository
      Parameters:
      task - the task to persist — 待持久化的任务
      Returns:
      the persisted task — 已持久化的任务

    • findTaskById

      public Optional<Task> findTaskById(String id)
      Description copied from interface: FlowRepository
      Looks up a task by id. 根据 ID 查找任务。
      Specified by:
      findTaskById in interface FlowRepository
      Parameters:
      id - task id — 任务 ID
      Returns:
      the task wrapped in an Optional — 任务（Optional）

    • findTasksByInstanceAndStatus

      public List<Task> findTasksByInstanceAndStatus(String instanceId, TaskStatus status)
      Description copied from interface: FlowRepository
      Lists tasks that belong to the given instance and currently sit in the given status. 列出指定实例下、指定状态的任务。
      Specified by:
      findTasksByInstanceAndStatus in interface FlowRepository
      Parameters:
      instanceId - owning instance id — 所属实例 ID
      status - task status filter — 任务状态
      Returns:
      matching tasks — 符合条件的任务

    • findTasksByInstance

      public List<Task> findTasksByInstance(String instanceId)
      Description copied from interface: FlowRepository
      Lists every task (any status) under the given instance. 列出指定实例下所有任务（不限状态）。
      Specified by:
      findTasksByInstance in interface FlowRepository
      Parameters:
      instanceId - owning instance id — 所属实例 ID
      Returns:
      all tasks — 所有任务

    • findTodoByUser

      public List<Task> findTodoByUser(String userId, int page, int size)
      Description copied from interface: FlowRepository
      Paginated todo-list query for the given user (status = TaskStatus.PENDING only). 分页查询某用户的待办（仅 TaskStatus.PENDING）。
      Specified by:
      findTodoByUser in interface FlowRepository
      Parameters:
      userId - assignee user id — 承接人 userId
      page - zero-based page index — 页码（0 起）
      size - page size — 每页大小
      Returns:
      tasks on the requested page — 该页任务

    • updateTask

      public int updateTask(Task task)
      Description copied from interface: FlowRepository
      Optimistic-lock UPDATE; returns affected rows (0 = version conflict). 乐观锁 UPDATE；返回 affected rows（0 表示版本冲突）。
      Specified by:
      updateTask in interface FlowRepository
      Parameters:
      task - the task to update — 待更新的任务
      Returns:
      affected rows — 影响行数

    • cancelOtherPendingTasks

      public int cancelOtherPendingTasks(String instanceId, String stepId, String exceptTaskId)
      Description copied from interface: FlowRepository
      Bulk-cancels every other TaskStatus.PENDING task on the same step (excluding exceptTaskId), respecting optimistic-lock version. Returns the number of tasks that were successfully canceled. 批量取消同节点其他 PENDING 任务（不含 exceptTaskId），遵守乐观锁； 返回成功取消的条数。
      Specified by:
      cancelOtherPendingTasks in interface FlowRepository
      Parameters:
      instanceId - owning instance id — 所属实例 ID
      stepId - step id — 节点 ID
      exceptTaskId - the task id to keep — 保留的任务 ID
      Returns:
      number of canceled tasks — 成功取消条数

    • cancelAllTasksForStep

      public int cancelAllTasksForStep(String instanceId, String stepId)
      Description copied from interface: FlowRepository
      Bulk-transitions every task under (instanceId, stepId) whose current status is not TaskStatus.CANCELED into the TaskStatus.CANCELED state — including DONE / REJECTED rows from a prior approval round. Used by RejectOperation when rolling back to a previously visited step so that StepEvaluator only ever observes tasks from the new batch. 批量将 (instanceId, stepId) 下所有非 CANCELED 状态的任务（含上一轮 的 DONE / REJECTED）置为 TaskStatus.CANCELED。 RejectOperation 在回退到已访问节点时调用此方法，确保 StepEvaluator 只看见新一轮批次的任务。

      Invariant: each step has at most one active batch of tasks at any given time. Reject re-entry must enforce this by calling this method for every affected step before creating fresh PENDING rows.

      不变式：任意时刻每个 step 最多有一批活跃 task；reject 重入前必须对所有 受影响 step 调用本方法。

      Specified by:
      cancelAllTasksForStep in interface FlowRepository
      Parameters:
      instanceId - owning instance id — 所属实例 ID
      stepId - step id — 节点 ID
      Returns:
      number of affected rows — 影响行数

    • appendHistory

      public void appendHistory(HistoryRecord record)
      Description copied from interface: FlowRepository
      Appends a new history row. History is append-only — once written, rows are never updated or deleted. 追加一条历史记录。历史表仅追加，写入后不可更新或删除。
      Specified by:
      appendHistory in interface FlowRepository
      Parameters:
      record - the history row to append — 待追加的历史

    • findHistoryByInstance

      public List<HistoryRecord> findHistoryByInstance(String instanceId)
      Description copied from interface: FlowRepository
      Lists every history row for an instance in chronological order (earliest first). 按时间升序列出指定实例的所有历史记录。
      Specified by:
      findHistoryByInstance in interface FlowRepository
      Parameters:
      instanceId - owning instance id — 所属实例 ID
      Returns:
      chronological history — 时间升序的历史

    • findHistoryByUser

      public List<HistoryRecord> findHistoryByUser(String userId, int page, int size)
      Description copied from interface: FlowRepository
      Paginated history query for the given user (matched on HistoryRecord.operatorId()). 分页查询某用户的历史记录（按 HistoryRecord.operatorId() 匹配）。
      Specified by:
      findHistoryByUser in interface FlowRepository
      Parameters:
      userId - operator user id — 操作人 userId
      page - zero-based page index — 页码（0 起）
      size - page size — 每页大小
      Returns:
      history rows on the requested page — 该页历史

    • findHistoryByTaskId

      public Optional<HistoryRecord> findHistoryByTaskId(String taskId)
      Description copied from interface: FlowRepository
      Looks up a history row by its related task id (one-to-one for terminal task actions such as approve / reject). 按关联任务 ID 查找历史（approve / reject 等终态任务动作一对一）。
      Specified by:
      findHistoryByTaskId in interface FlowRepository
      Parameters:
      taskId - related task id — 关联任务 ID
      Returns:
      the history row wrapped in an Optional — 历史（Optional）

    • recordVote

      public void recordVote(String taskId, String voterId, VoteOpinion opinion)
      Description copied from interface: FlowRepository
      Records a single vote at a StepType.VOTE step. Append-only. 在 StepType.VOTE 节点记录一票。追加写。
      Specified by:
      recordVote in interface FlowRepository
      Parameters:
      taskId - related task id — 关联任务 ID
      voterId - voter user id — 投票人
      opinion - vote opinion — 投票意见

    • findVotesByTask

      public List<VoteRecord> findVotesByTask(String taskId)
      Description copied from interface: FlowRepository
      Lists every vote recorded against the given task, in vote order. 列出关联任务的所有投票（按投票顺序）。
      Specified by:
      findVotesByTask in interface FlowRepository
      Parameters:
      taskId - related task id — 关联任务 ID
      Returns:
      votes in submission order — 投票（按提交顺序）

    • runInTransaction

      public void runInTransaction(Runnable action)
      Runs the supplied action atomically (R1 S-5).
      • Re-uses an outer transaction when nested.
      • Opens a single connection with autoCommit=false, binds it to this thread, runs the action, then commits on success or rolls back on exception.
      原子执行动作（R1 S-5）；嵌套调用自动并入外层； 取一个 autoCommit=false 连接，提交或回滚。
      Specified by:
      runInTransaction in interface FlowRepository
      Parameters:
      action - the action — 待执行动作
      Since:
      JDK 25, opencode-base-flow V1.0.5