mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(workflow): add foundational schema for workflow runtime
This commit is contained in:
matevip
parent
91e93e8831
commit
d87ed0365a
@ -0,0 +1,71 @@
|
||||
package vip.mate.trigger.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldFill;
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableLogic;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Trigger row. {@code patternVersion} is a lamport counter that fire callbacks
|
||||
* compare against the row on every fire; mismatches mean another instance has
|
||||
* updated the cron expression and the local schedule must self-cancel.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_trigger")
|
||||
public class TriggerEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long workspaceId;
|
||||
|
||||
@TableField(value = "name", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String name;
|
||||
|
||||
/** Pattern flavour: cron / webhook / channel_message / agent_lifecycle / content_match / workflow_completion. */
|
||||
private String patternType;
|
||||
|
||||
@TableField(value = "pattern_json", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String patternJson;
|
||||
|
||||
/** Routing target type: agent or workflow. */
|
||||
private String targetType;
|
||||
|
||||
private Long targetId;
|
||||
|
||||
@TableField(value = "payload_template", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String payloadTemplate;
|
||||
|
||||
private Integer rateLimitPerMin;
|
||||
|
||||
private Integer dedupWindowSecs;
|
||||
|
||||
private Boolean botSelfFilter;
|
||||
|
||||
private Boolean enabled;
|
||||
|
||||
private Long fireCount;
|
||||
|
||||
private Long maxFires;
|
||||
|
||||
@TableField(value = "last_fired_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime lastFiredAt;
|
||||
|
||||
/** Lamport counter — bump on every cron expression / payload template change. */
|
||||
private Long patternVersion;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private LocalDateTime createTime;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT_UPDATE)
|
||||
private LocalDateTime updateTime;
|
||||
|
||||
@TableLogic
|
||||
private Integer deleted;
|
||||
}
|
||||
@ -0,0 +1,31 @@
|
||||
package vip.mate.trigger.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Trigger dedup-window row. {@code dedupKey} carries envelope.eventId, falling
|
||||
* back to a content sha256 when the upstream channel did not provide a stable
|
||||
* id. {@code expiresAt} is set on insert to {@code receivedAt + dedupWindowSecs}
|
||||
* so the cleanup task can sweep expired rows.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_trigger_event")
|
||||
public class TriggerEventEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long triggerId;
|
||||
|
||||
private String dedupKey;
|
||||
|
||||
/** Filled by DB DEFAULT CURRENT_TIMESTAMP when left null on insert. */
|
||||
private LocalDateTime receivedAt;
|
||||
|
||||
private LocalDateTime expiresAt;
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.trigger.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.trigger.model.TriggerEventEntity;
|
||||
|
||||
@Mapper
|
||||
public interface TriggerEventMapper extends BaseMapper<TriggerEventEntity> {
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.trigger.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.trigger.model.TriggerEntity;
|
||||
|
||||
@Mapper
|
||||
public interface TriggerMapper extends BaseMapper<TriggerEntity> {
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
package vip.mate.workflow.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldFill;
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableLogic;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Stable workflow identity. The current draft is stored inline (1:1 with the
|
||||
* workflow row) so PK uniqueness automatically guarantees a single draft;
|
||||
* published snapshots live in {@code mate_workflow_revision}.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_workflow")
|
||||
public class WorkflowEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long workspaceId;
|
||||
|
||||
private String name;
|
||||
|
||||
private String description;
|
||||
|
||||
private Boolean enabled;
|
||||
|
||||
/** Inline draft graph_json; null when there is no active draft. */
|
||||
@TableField(value = "draft_json", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String draftJson;
|
||||
|
||||
@TableField(value = "draft_schema_version", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String draftSchemaVersion;
|
||||
|
||||
@TableField(value = "draft_updated_by", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long draftUpdatedBy;
|
||||
|
||||
@TableField(value = "draft_updated_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime draftUpdatedAt;
|
||||
|
||||
/** Pointer to the most recently published revision; null if never published. */
|
||||
@TableField(value = "latest_revision_id", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long latestRevisionId;
|
||||
|
||||
private Long createdBy;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private LocalDateTime createTime;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT_UPDATE)
|
||||
private LocalDateTime updateTime;
|
||||
|
||||
@TableLogic
|
||||
private Integer deleted;
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
package vip.mate.workflow.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Payload body addressed by a stable URI. Small payloads (< 256KB) live
|
||||
* inline in {@code contentBytes}; larger payloads point at filesystem or
|
||||
* object storage via {@code storageKind} + {@code storageRef}. {@code sha256}
|
||||
* is for tamper detection only — v0 does not deduplicate across runs.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_workflow_payload")
|
||||
public class WorkflowPayloadEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private String payloadUri;
|
||||
|
||||
private Long workspaceId;
|
||||
|
||||
@TableField(value = "content_bytes", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private byte[] contentBytes;
|
||||
|
||||
/** Storage flavour: inline / fs / s3 / oss. */
|
||||
private String storageKind;
|
||||
|
||||
@TableField(value = "storage_ref", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String storageRef;
|
||||
|
||||
@TableField(value = "content_type", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String contentType;
|
||||
|
||||
@TableField(value = "sha256", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String sha256;
|
||||
|
||||
@TableField(value = "size_bytes", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long sizeBytes;
|
||||
|
||||
private LocalDateTime createdAt;
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
package vip.mate.workflow.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldFill;
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Immutable published snapshot of a workflow. The {@code revision} column is
|
||||
* monotonic per workflow; rows are append-only after publish.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_workflow_revision")
|
||||
public class WorkflowRevisionEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long workflowId;
|
||||
|
||||
private Integer revision;
|
||||
|
||||
@TableField(value = "graph_json", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String graphJson;
|
||||
|
||||
private String schemaVersion;
|
||||
|
||||
@TableField(value = "published_note", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String publishedNote;
|
||||
|
||||
@TableField(value = "published_by", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long publishedBy;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private LocalDateTime createTime;
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
package vip.mate.workflow.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldFill;
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableLogic;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Workflow run instance. Run is locked to a specific revision for stability
|
||||
* even when later revisions are published. Initial input and final output are
|
||||
* stored as payload URIs to avoid bloating the run row.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_workflow_run")
|
||||
public class WorkflowRunEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long workflowId;
|
||||
|
||||
private Long revisionId;
|
||||
|
||||
private Long workspaceId;
|
||||
|
||||
/** State machine value: pending / running / paused / succeeded / failed / cancelled / timed_out. */
|
||||
private String state;
|
||||
|
||||
@TableField(value = "triggered_by", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String triggeredBy;
|
||||
|
||||
@TableField(value = "triggered_meta", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String triggeredMeta;
|
||||
|
||||
@TableField(value = "initial_input_ref", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String initialInputRef;
|
||||
|
||||
@TableField(value = "final_output_ref", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String finalOutputRef;
|
||||
|
||||
@TableField(value = "error_message", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String errorMessage;
|
||||
|
||||
@TableField(value = "started_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime startedAt;
|
||||
|
||||
@TableField(value = "completed_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime completedAt;
|
||||
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private LocalDateTime createTime;
|
||||
|
||||
@TableLogic
|
||||
private Integer deleted;
|
||||
}
|
||||
@ -0,0 +1,51 @@
|
||||
package vip.mate.workflow.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Durable workflow pause row. Holds the resume token and links back to the
|
||||
* external approval row (or other callback source) so that resume can be
|
||||
* triggered idempotently after a JVM restart.
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_workflow_run_pause")
|
||||
public class WorkflowRunPauseEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long runId;
|
||||
|
||||
private Long stepId;
|
||||
|
||||
/** Source of the pause: await_approval, external_callback, etc. */
|
||||
private String pauseKind;
|
||||
|
||||
/** Random server-generated token used as the resume entry key. */
|
||||
private String pauseToken;
|
||||
|
||||
@TableField(value = "external_approval_id", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long externalApprovalId;
|
||||
|
||||
private LocalDateTime pausedAt;
|
||||
|
||||
@TableField(value = "resume_deadline", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime resumeDeadline;
|
||||
|
||||
@TableField(value = "resume_payload_ref", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String resumePayloadRef;
|
||||
|
||||
@TableField(value = "resumed_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime resumedAt;
|
||||
|
||||
/** Outcome on resume: approved / rejected / timeout / cancelled. */
|
||||
@TableField(value = "resume_outcome", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String resumeOutcome;
|
||||
}
|
||||
@ -0,0 +1,68 @@
|
||||
package vip.mate.workflow.model;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.FieldStrategy;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* Per-step execution row. {@code stepIndex} is the zero-based index in the
|
||||
* revision's steps array; {@code iterationIndex} is reserved for fan_out
|
||||
* iterations (and future loop bodies).
|
||||
*/
|
||||
@Data
|
||||
@TableName("mate_workflow_run_step")
|
||||
public class WorkflowRunStepEntity {
|
||||
|
||||
@TableId(type = IdType.ASSIGN_ID)
|
||||
private Long id;
|
||||
|
||||
private Long runId;
|
||||
|
||||
private Integer stepIndex;
|
||||
|
||||
@TableField(value = "iteration_index", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Integer iterationIndex;
|
||||
|
||||
@TableField(value = "step_name", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String stepName;
|
||||
|
||||
@TableField(value = "agent_id", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long agentId;
|
||||
|
||||
private String state;
|
||||
|
||||
@TableField(value = "input_ref", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String inputRef;
|
||||
|
||||
@TableField(value = "output_ref", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String outputRef;
|
||||
|
||||
@TableField(value = "output_summary", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String outputSummary;
|
||||
|
||||
@TableField(value = "output_content_type", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String outputContentType;
|
||||
|
||||
@TableField(value = "error_message", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private String errorMessage;
|
||||
|
||||
@TableField(value = "duration_ms", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Long durationMs;
|
||||
|
||||
@TableField(value = "token_input", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Integer tokenInput;
|
||||
|
||||
@TableField(value = "token_output", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private Integer tokenOutput;
|
||||
|
||||
@TableField(value = "started_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime startedAt;
|
||||
|
||||
@TableField(value = "completed_at", updateStrategy = FieldStrategy.ALWAYS)
|
||||
private LocalDateTime completedAt;
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.workflow.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.workflow.model.WorkflowEntity;
|
||||
|
||||
@Mapper
|
||||
public interface WorkflowMapper extends BaseMapper<WorkflowEntity> {
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.workflow.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.workflow.model.WorkflowPayloadEntity;
|
||||
|
||||
@Mapper
|
||||
public interface WorkflowPayloadMapper extends BaseMapper<WorkflowPayloadEntity> {
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.workflow.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.workflow.model.WorkflowRevisionEntity;
|
||||
|
||||
@Mapper
|
||||
public interface WorkflowRevisionMapper extends BaseMapper<WorkflowRevisionEntity> {
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.workflow.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.workflow.model.WorkflowRunEntity;
|
||||
|
||||
@Mapper
|
||||
public interface WorkflowRunMapper extends BaseMapper<WorkflowRunEntity> {
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.workflow.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.workflow.model.WorkflowRunPauseEntity;
|
||||
|
||||
@Mapper
|
||||
public interface WorkflowRunPauseMapper extends BaseMapper<WorkflowRunPauseEntity> {
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package vip.mate.workflow.repository;
|
||||
|
||||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import vip.mate.workflow.model.WorkflowRunStepEntity;
|
||||
|
||||
@Mapper
|
||||
public interface WorkflowRunStepMapper extends BaseMapper<WorkflowRunStepEntity> {
|
||||
}
|
||||
@ -0,0 +1,173 @@
|
||||
-- V96: Foundational schema for the workflow runtime.
|
||||
-- Eight tables establish workflow identity (workflow + immutable revisions),
|
||||
-- run state (run + per-step rows + durable pause rows for await_approval),
|
||||
-- payload URI storage with inline / filesystem fallback, and trigger
|
||||
-- definitions paired with a dedup-window table for envelope-based event
|
||||
-- governance. H2 dialect uses CLOB for MEDIUMTEXT and BLOB for LONGBLOB;
|
||||
-- secondary indexes are emitted as separate CREATE INDEX statements per
|
||||
-- project convention.
|
||||
|
||||
-- 1. Stable workflow identity + draft (1:1 with workflow row).
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
name VARCHAR(128) NOT NULL,
|
||||
description VARCHAR(1024),
|
||||
enabled BOOLEAN NOT NULL DEFAULT TRUE,
|
||||
draft_json CLOB,
|
||||
draft_schema_version VARCHAR(8),
|
||||
draft_updated_by BIGINT,
|
||||
draft_updated_at TIMESTAMP,
|
||||
latest_revision_id BIGINT,
|
||||
created_by BIGINT,
|
||||
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
deleted INT NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_workflow_workspace_name
|
||||
ON mate_workflow (workspace_id, name, deleted);
|
||||
|
||||
-- 2. Immutable published revisions; integer revision is monotonic per workflow.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_revision (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workflow_id BIGINT NOT NULL,
|
||||
revision INT NOT NULL,
|
||||
graph_json CLOB NOT NULL,
|
||||
schema_version VARCHAR(8) NOT NULL,
|
||||
published_note VARCHAR(512),
|
||||
published_by BIGINT,
|
||||
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_workflow_revision
|
||||
ON mate_workflow_revision (workflow_id, revision);
|
||||
|
||||
-- 3. Workflow run instance; payload bodies live behind URIs in mate_workflow_payload.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_run (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workflow_id BIGINT NOT NULL,
|
||||
revision_id BIGINT NOT NULL,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
state VARCHAR(16) NOT NULL,
|
||||
triggered_by VARCHAR(32),
|
||||
triggered_meta CLOB,
|
||||
initial_input_ref VARCHAR(256),
|
||||
final_output_ref VARCHAR(256),
|
||||
error_message VARCHAR(2048),
|
||||
started_at TIMESTAMP,
|
||||
completed_at TIMESTAMP,
|
||||
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
deleted INT NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_workflow_run_started
|
||||
ON mate_workflow_run (workflow_id, started_at);
|
||||
|
||||
-- 4. Per-step run row; iteration_index reserved for fan_out (and future loop).
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_run_step (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
run_id BIGINT NOT NULL,
|
||||
step_index INT NOT NULL,
|
||||
iteration_index INT,
|
||||
step_name VARCHAR(128),
|
||||
agent_id BIGINT,
|
||||
state VARCHAR(16),
|
||||
input_ref VARCHAR(256),
|
||||
output_ref VARCHAR(256),
|
||||
output_summary VARCHAR(512),
|
||||
output_content_type VARCHAR(64),
|
||||
error_message VARCHAR(2048),
|
||||
duration_ms BIGINT,
|
||||
token_input INT,
|
||||
token_output INT,
|
||||
started_at TIMESTAMP,
|
||||
completed_at TIMESTAMP
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_workflow_run_step
|
||||
ON mate_workflow_run_step (run_id, step_index, iteration_index);
|
||||
|
||||
-- 5. Durable pause rows so await_approval can resume across restarts.
|
||||
-- pause_token is the resume entry key; external_approval_id ties back to
|
||||
-- ApprovalWorkflowService rows so the approval callback can find the pause.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_run_pause (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
run_id BIGINT NOT NULL,
|
||||
step_id BIGINT NOT NULL,
|
||||
pause_kind VARCHAR(32) NOT NULL,
|
||||
pause_token VARCHAR(128) NOT NULL,
|
||||
external_approval_id BIGINT,
|
||||
paused_at TIMESTAMP NOT NULL,
|
||||
resume_deadline TIMESTAMP,
|
||||
resume_payload_ref VARCHAR(256),
|
||||
resumed_at TIMESTAMP,
|
||||
resume_outcome VARCHAR(32)
|
||||
);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_workflow_pause_run_step
|
||||
ON mate_workflow_run_pause (run_id, step_id);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_workflow_pause_token
|
||||
ON mate_workflow_run_pause (pause_token);
|
||||
CREATE INDEX IF NOT EXISTS idx_workflow_pause_external_approval
|
||||
ON mate_workflow_run_pause (external_approval_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_workflow_pause_open_deadline
|
||||
ON mate_workflow_run_pause (resumed_at, resume_deadline);
|
||||
|
||||
-- 6. Payload URI storage. Inline blob for < 256KB; storage_kind=fs/s3/oss
|
||||
-- carries the external object key in storage_ref. sha256 is for tamper
|
||||
-- detection only — v0 does not deduplicate across runs.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_payload (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
payload_uri VARCHAR(256) NOT NULL,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
content_bytes BLOB,
|
||||
storage_kind VARCHAR(16) NOT NULL,
|
||||
storage_ref VARCHAR(512),
|
||||
content_type VARCHAR(64),
|
||||
sha256 CHAR(64),
|
||||
size_bytes BIGINT,
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_workflow_payload_uri
|
||||
ON mate_workflow_payload (payload_uri);
|
||||
CREATE INDEX IF NOT EXISTS idx_workflow_payload_workspace_created
|
||||
ON mate_workflow_payload (workspace_id, created_at);
|
||||
|
||||
-- 7. Trigger definitions. pattern_version is a lamport counter that fire
|
||||
-- callbacks compare against on every fire to detect that another instance
|
||||
-- has updated the cron expression and self-cancel the local schedule.
|
||||
CREATE TABLE IF NOT EXISTS mate_trigger (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
name VARCHAR(128),
|
||||
pattern_type VARCHAR(32) NOT NULL,
|
||||
pattern_json CLOB NOT NULL,
|
||||
target_type VARCHAR(16) NOT NULL,
|
||||
target_id BIGINT NOT NULL,
|
||||
payload_template CLOB,
|
||||
rate_limit_per_min INT NOT NULL DEFAULT 60,
|
||||
dedup_window_secs INT NOT NULL DEFAULT 60,
|
||||
bot_self_filter BOOLEAN NOT NULL DEFAULT TRUE,
|
||||
enabled BOOLEAN NOT NULL DEFAULT TRUE,
|
||||
fire_count BIGINT NOT NULL DEFAULT 0,
|
||||
max_fires BIGINT NOT NULL DEFAULT 0,
|
||||
last_fired_at TIMESTAMP,
|
||||
pattern_version BIGINT NOT NULL DEFAULT 1,
|
||||
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
deleted INT NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_trigger_workspace_enabled
|
||||
ON mate_trigger (workspace_id, enabled, deleted);
|
||||
CREATE INDEX IF NOT EXISTS idx_trigger_target
|
||||
ON mate_trigger (target_type, target_id);
|
||||
|
||||
-- 8. Event dedup window. dedup_key is envelope.eventId, falling back to
|
||||
-- sourceHash when the upstream channel did not provide a stable id.
|
||||
CREATE TABLE IF NOT EXISTS mate_trigger_event (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
trigger_id BIGINT NOT NULL,
|
||||
dedup_key VARCHAR(128) NOT NULL,
|
||||
received_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
expires_at TIMESTAMP NOT NULL
|
||||
);
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS uk_trigger_dedup
|
||||
ON mate_trigger_event (trigger_id, dedup_key);
|
||||
CREATE INDEX IF NOT EXISTS idx_trigger_event_expires
|
||||
ON mate_trigger_event (expires_at);
|
||||
@ -0,0 +1,162 @@
|
||||
-- V96: Foundational schema for the workflow runtime.
|
||||
-- Eight tables establish workflow identity (workflow + immutable revisions),
|
||||
-- run state (run + per-step rows + durable pause rows for await_approval),
|
||||
-- payload URI storage with inline / filesystem fallback, and trigger
|
||||
-- definitions paired with a dedup-window table for envelope-based event
|
||||
-- governance. CREATE TABLE IF NOT EXISTS is itself idempotent on MySQL.
|
||||
|
||||
-- 1. Stable workflow identity + draft (1:1 with workflow row).
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
name VARCHAR(128) NOT NULL,
|
||||
description VARCHAR(1024),
|
||||
enabled TINYINT NOT NULL DEFAULT 1,
|
||||
draft_json MEDIUMTEXT,
|
||||
draft_schema_version VARCHAR(8),
|
||||
draft_updated_by BIGINT,
|
||||
draft_updated_at DATETIME(3),
|
||||
latest_revision_id BIGINT,
|
||||
created_by BIGINT,
|
||||
create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
update_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
|
||||
deleted INT NOT NULL DEFAULT 0,
|
||||
UNIQUE KEY uk_workflow_workspace_name (workspace_id, name, deleted)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Workflow definition with stable identity and inline draft.';
|
||||
|
||||
-- 2. Immutable published revisions; integer revision is monotonic per workflow.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_revision (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workflow_id BIGINT NOT NULL,
|
||||
revision INT NOT NULL,
|
||||
graph_json MEDIUMTEXT NOT NULL,
|
||||
schema_version VARCHAR(8) NOT NULL,
|
||||
published_note VARCHAR(512),
|
||||
published_by BIGINT,
|
||||
create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
UNIQUE KEY uk_workflow_revision (workflow_id, revision)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Immutable published workflow revisions.';
|
||||
|
||||
-- 3. Workflow run instance; payload bodies live behind URIs in mate_workflow_payload.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_run (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workflow_id BIGINT NOT NULL,
|
||||
revision_id BIGINT NOT NULL,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
state VARCHAR(16) NOT NULL,
|
||||
triggered_by VARCHAR(32),
|
||||
triggered_meta MEDIUMTEXT,
|
||||
initial_input_ref VARCHAR(256),
|
||||
final_output_ref VARCHAR(256),
|
||||
error_message VARCHAR(2048),
|
||||
started_at DATETIME(3),
|
||||
completed_at DATETIME(3),
|
||||
create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
deleted INT NOT NULL DEFAULT 0,
|
||||
KEY idx_workflow_run_started (workflow_id, started_at)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Workflow run instances locked to a specific revision.';
|
||||
|
||||
-- 4. Per-step run row; iteration_index reserved for fan_out (and future loop).
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_run_step (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
run_id BIGINT NOT NULL,
|
||||
step_index INT NOT NULL,
|
||||
iteration_index INT,
|
||||
step_name VARCHAR(128),
|
||||
agent_id BIGINT,
|
||||
state VARCHAR(16),
|
||||
input_ref VARCHAR(256),
|
||||
output_ref VARCHAR(256),
|
||||
output_summary VARCHAR(512),
|
||||
output_content_type VARCHAR(64),
|
||||
error_message VARCHAR(2048),
|
||||
duration_ms BIGINT,
|
||||
token_input INT,
|
||||
token_output INT,
|
||||
started_at DATETIME(3),
|
||||
completed_at DATETIME(3),
|
||||
KEY idx_workflow_run_step (run_id, step_index, iteration_index)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Per-step run rows with input/output references and timings.';
|
||||
|
||||
-- 5. Durable pause rows so await_approval can resume across restarts.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_run_pause (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
run_id BIGINT NOT NULL,
|
||||
step_id BIGINT NOT NULL,
|
||||
pause_kind VARCHAR(32) NOT NULL,
|
||||
pause_token VARCHAR(128) NOT NULL,
|
||||
external_approval_id BIGINT,
|
||||
paused_at DATETIME(3) NOT NULL,
|
||||
resume_deadline DATETIME(3),
|
||||
resume_payload_ref VARCHAR(256),
|
||||
resumed_at DATETIME(3),
|
||||
resume_outcome VARCHAR(32),
|
||||
UNIQUE KEY uk_workflow_pause_run_step (run_id, step_id),
|
||||
UNIQUE KEY uk_workflow_pause_token (pause_token),
|
||||
KEY idx_workflow_pause_external_approval (external_approval_id),
|
||||
KEY idx_workflow_pause_open_deadline (resumed_at, resume_deadline)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Durable workflow pause rows for await_approval resume.';
|
||||
|
||||
-- 6. Payload URI storage. Inline blob for < 256KB; storage_kind=fs/s3/oss
|
||||
-- carries the external object key in storage_ref.
|
||||
CREATE TABLE IF NOT EXISTS mate_workflow_payload (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
payload_uri VARCHAR(256) NOT NULL,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
content_bytes LONGBLOB,
|
||||
storage_kind VARCHAR(16) NOT NULL,
|
||||
storage_ref VARCHAR(512),
|
||||
content_type VARCHAR(64),
|
||||
sha256 CHAR(64),
|
||||
size_bytes BIGINT,
|
||||
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
UNIQUE KEY uk_workflow_payload_uri (payload_uri),
|
||||
KEY idx_workflow_payload_workspace_created (workspace_id, created_at)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Payload bodies addressed by stable URIs.';
|
||||
|
||||
-- 7. Trigger definitions. pattern_version is a lamport counter that fire
|
||||
-- callbacks compare against on every fire to detect that another instance
|
||||
-- has updated the cron expression and self-cancel the local schedule.
|
||||
CREATE TABLE IF NOT EXISTS mate_trigger (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
workspace_id BIGINT NOT NULL,
|
||||
name VARCHAR(128),
|
||||
pattern_type VARCHAR(32) NOT NULL,
|
||||
pattern_json MEDIUMTEXT NOT NULL,
|
||||
target_type VARCHAR(16) NOT NULL,
|
||||
target_id BIGINT NOT NULL,
|
||||
payload_template MEDIUMTEXT,
|
||||
rate_limit_per_min INT NOT NULL DEFAULT 60,
|
||||
dedup_window_secs INT NOT NULL DEFAULT 60,
|
||||
bot_self_filter TINYINT NOT NULL DEFAULT 1,
|
||||
enabled TINYINT NOT NULL DEFAULT 1,
|
||||
fire_count BIGINT NOT NULL DEFAULT 0,
|
||||
max_fires BIGINT NOT NULL DEFAULT 0,
|
||||
last_fired_at DATETIME(3),
|
||||
pattern_version BIGINT NOT NULL DEFAULT 1,
|
||||
create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
update_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
|
||||
deleted INT NOT NULL DEFAULT 0,
|
||||
KEY idx_trigger_workspace_enabled (workspace_id, enabled, deleted),
|
||||
KEY idx_trigger_target (target_type, target_id)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Workflow / agent trigger definitions with pattern versioning.';
|
||||
|
||||
-- 8. Event dedup window. dedup_key is envelope.eventId, falling back to
|
||||
-- sourceHash when the upstream channel did not provide a stable id.
|
||||
CREATE TABLE IF NOT EXISTS mate_trigger_event (
|
||||
id BIGINT NOT NULL PRIMARY KEY,
|
||||
trigger_id BIGINT NOT NULL,
|
||||
dedup_key VARCHAR(128) NOT NULL,
|
||||
received_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
expires_at DATETIME(3) NOT NULL,
|
||||
UNIQUE KEY uk_trigger_dedup (trigger_id, dedup_key),
|
||||
KEY idx_trigger_event_expires (expires_at)
|
||||
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci
|
||||
COMMENT = 'Per-trigger event dedup window with TTL-style expiry.';
|
||||
Loading…
Reference in New Issue
Block a user