126 lines
7.8 KiB
Go
126 lines
7.8 KiB
Go
// Package instance 定义流程实例的受理与查询边界；执行推进由持久化存储运行器负责。
|
||
package instance
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"strings"
|
||
"time"
|
||
"unicode/utf8"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
)
|
||
|
||
const StartAction = "workflows:start"
|
||
const ReadAction = "workflows:read"
|
||
const MaxUnfinished = 10000
|
||
|
||
// StartInput 使用显式版本；业务追踪键不能替代请求幂等键。
|
||
type StartInput struct {
|
||
DefinitionID core.ID `json:"definition_id"` // DefinitionID 同命名空间的稳定定义身份。
|
||
Version string `json:"version"` // Version 已发布的固定版本名，不解析latest别名。
|
||
Input json.RawMessage `json:"input" nullable:"true"` // Input 按固定版本验证的输入，显式null与未提供不同。
|
||
BusinessKey string `json:"business_key,omitempty"` // BusinessKey 最多191个Unicode字符，保留尾部空格。
|
||
}
|
||
|
||
// View 是实例当前权威视图；StartResult中的View冻结在受理事务，不随重报更新。
|
||
type View struct {
|
||
ID core.ID `json:"id"` // ID 实例身份。
|
||
NamespaceID core.ID `json:"namespace_id"` // NamespaceID 授权及数据隔离范围。
|
||
DefinitionID core.ID `json:"definition_id"` // DefinitionID 稳定定义身份。
|
||
DefinitionVersionID core.ID `json:"definition_version_id"` // DefinitionVersionID 固定版本记录身份。
|
||
Version string `json:"version"` // Version 启动时固定的发布版本名。
|
||
State weave.InstanceState `json:"state"` // State 执行状态，受理不代表业务成功。
|
||
Revision uint64 `json:"revision,string"` // Revision 实例并发版本。
|
||
Paused bool `json:"paused"` // Paused 独立暂停标志，不表示在途工作已停止。
|
||
ControlRevision uint64 `json:"control_revision,string"` // ControlRevision 控制操作并发版本。
|
||
Input json.RawMessage `json:"input"` // Input 不变的受理输入，保留数字文本。
|
||
Output json.RawMessage `json:"output,omitempty"` // Output 成功输出；缺失与显式null不同。
|
||
BusinessKey string `json:"business_key"` // BusinessKey 原始追踪键。
|
||
DeadlineAt time.Time `json:"deadline_at"` // DeadlineAt 数据库时间确定的固定总截止。
|
||
CancelRequestedAt *time.Time `json:"cancel_requested_at,omitempty"` // CancelRequestedAt 已保存的取消意图时间。
|
||
StopKind *string `json:"stop_kind,omitempty"` // StopKind 首次停止原因CANCEL或FAILURE；核查不会清除。
|
||
StopRequestedAt *time.Time `json:"stop_requested_at,omitempty"` // StopRequestedAt 首次停止意图时间，后续操作不会延长。
|
||
CompletedAt *time.Time `json:"completed_at,omitempty"` // CompletedAt 终结时间，非终态为空。
|
||
CreatedAt time.Time `json:"created_at"` // CreatedAt UTC受理时间。
|
||
UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最近状态更新时间。
|
||
}
|
||
type StartResult struct {
|
||
Instance View `json:"instance"` // Instance 原受理回执的实例快照。
|
||
Deduplicated bool `json:"deduplicated"` // Deduplicated 原键恢复，不新建工作或续期。
|
||
}
|
||
|
||
// StepView 只列元数据，避免把最多128个大输入输出聚合为无界响应。
|
||
type StepView struct {
|
||
ActivationSequence uint32 `json:"activation_sequence,omitempty"` // ActivationSequence 当前激活代次；未激活时省略，信号必须显式指定此代次。
|
||
ID core.ID `json:"id"` // ID 实例内逻辑步骤身份。
|
||
Key string `json:"key"` // Key 固定定义步骤名称。
|
||
Kind weave.NodeKind `json:"kind"` // Kind 固定节点类型。
|
||
State weave.StepState `json:"state"` // State 当前步骤状态。
|
||
Revision uint64 `json:"revision,string"` // Revision 步骤并发版本。
|
||
ActivationID *core.ID `json:"activation_id,omitempty"` // ActivationID 当前激活，未激活时为空。
|
||
TaskID *core.ID `json:"task_id,omitempty"` // TaskID TASK的直接任务；POLL任务通过轮次查询。
|
||
}
|
||
type StepsView struct {
|
||
InstanceID core.ID `json:"instance_id"` // InstanceID 所属实例。
|
||
Revision uint64 `json:"revision,string"` // Revision 与步骤读取一致的实例版本。
|
||
Steps []StepView `json:"steps"` // Steps 按键排序，最多128个，不含内部根检查点。
|
||
}
|
||
type Store interface {
|
||
ListEvents(context.Context, auth.Token, core.ID, service.HistoryPageInput) (EventPage, error)
|
||
GetEvent(context.Context, auth.Token, core.ID, core.ID) (EventDetail, error)
|
||
List(context.Context, auth.Token, ListInput) (Page, error)
|
||
ListActivations(context.Context, auth.Token, core.ID, string, service.HistoryPageInput) (ActivationPage, error)
|
||
GetActivation(context.Context, auth.Token, core.ID, string, core.ID) (ActivationDetail, error)
|
||
ListPollRounds(context.Context, auth.Token, core.ID, string, core.ID, service.HistoryPageInput) (PollRoundPage, error)
|
||
ListOperations(context.Context, auth.Token, core.ID, service.HistoryPageInput) (OperationPage, error)
|
||
GetOperation(context.Context, auth.Token, core.ID, core.ID) (OperationView, error)
|
||
Start(context.Context, auth.Token, string, StartInput) (StartResult, error)
|
||
Get(context.Context, auth.Token, core.ID) (View, error)
|
||
Steps(context.Context, auth.Token, core.ID) (StepsView, error)
|
||
Step(context.Context, auth.Token, core.ID, string) (StepDetail, error)
|
||
}
|
||
type Service struct{ store Store }
|
||
|
||
func New(store Store) *Service { return &Service{store: store} }
|
||
func (s *Service) Start(ctx context.Context, t auth.Token, key string, in StartInput) (StartResult, error) {
|
||
if err := service.ValidateKey(key); err != nil {
|
||
return StartResult{}, err
|
||
}
|
||
if in.DefinitionID.Validate() != nil || !versionName(in.Version) || !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsRune(in.BusinessKey, 0) {
|
||
return StartResult{}, fault.Invalid("valid workflow definition, version and business key required")
|
||
}
|
||
if err := definition.ValidateData(definition.Type{Kind: "JSON"}, in.Input); err != nil {
|
||
return StartResult{}, fault.Invalid("valid workflow input required, maximum 256 KiB")
|
||
}
|
||
return s.store.Start(ctx, t, key, in)
|
||
}
|
||
func versionName(v string) bool {
|
||
if len(v) == 0 || len(v) > 64 {
|
||
return false
|
||
}
|
||
for i, r := range v {
|
||
if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || i > 0 && (r == '_' || r == '.' || r == '-')) {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
func (s *Service) Get(ctx context.Context, t auth.Token, id core.ID) (View, error) {
|
||
if id.Validate() != nil {
|
||
return View{}, fault.Invalid("invalid workflow instance id")
|
||
}
|
||
return s.store.Get(ctx, t, id)
|
||
}
|
||
func (s *Service) Steps(ctx context.Context, t auth.Token, id core.ID) (StepsView, error) {
|
||
if id.Validate() != nil {
|
||
return StepsView{}, fault.Invalid("invalid workflow instance id")
|
||
}
|
||
return s.store.Steps(ctx, t, id)
|
||
}
|