feat: add scoped workflow instance queries and SDK pagination
This commit is contained in:
257
internal/storage/mysql/workflow_query.go
Normal file
257
internal/storage/mysql/workflow_query.go
Normal file
@@ -0,0 +1,257 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
||||
)
|
||||
|
||||
// 查询游标只表示位置，不授予权限；改变权限范围必须重新查询。
|
||||
type workflowCursor struct {
|
||||
Version int `json:"v"`
|
||||
Namespace core.ID `json:"n"`
|
||||
Created time.Time `json:"t"`
|
||||
ID core.ID `json:"i"`
|
||||
Filter string `json:"f"`
|
||||
}
|
||||
|
||||
func workflowQueryQueues(p auth.Permissions) []string {
|
||||
if slices.Contains(p.Queues, "*") {
|
||||
return []string{"*"}
|
||||
}
|
||||
queues := slices.Clone(p.Queues)
|
||||
slices.Sort(queues)
|
||||
return slices.Compact(queues)
|
||||
}
|
||||
func workflowFilterDigest(in instance.ListInput, queues []string) (string, error) {
|
||||
// 页长和位置不属于筛选；nil暂停与false有不同语义。
|
||||
raw, err := requestDigest(struct {
|
||||
Domain string
|
||||
Definition core.ID
|
||||
Version string
|
||||
State string
|
||||
Paused *bool
|
||||
Business string
|
||||
From, Before *time.Time
|
||||
Queues []string
|
||||
}{"orbit.workflow-query.v1", in.DefinitionID, in.Version, string(in.State), in.Paused, in.BusinessKey, in.CreatedFrom, in.CreatedBefore, queues})
|
||||
return hex.EncodeToString(raw[:]), err
|
||||
}
|
||||
func decodeWorkflowCursor(value string, ns core.ID, in instance.ListInput, filter string) (workflowCursor, error) {
|
||||
var c workflowCursor
|
||||
invalid := fault.Invalid("cursor does not match this workflow query or queue permissions")
|
||||
raw, err := base64.RawURLEncoding.Strict().DecodeString(value)
|
||||
if err != nil || len(raw) > 768 {
|
||||
return c, invalid
|
||||
}
|
||||
if _, err = jsonvalue.Canonical(raw); err != nil {
|
||||
return c, invalid
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
if json.Unmarshal(raw, &fields) != nil || len(fields) != 5 {
|
||||
return c, invalid
|
||||
}
|
||||
for _, k := range []string{"v", "n", "t", "i", "f"} {
|
||||
v, ok := fields[k]
|
||||
if !ok || string(v) == "null" {
|
||||
return c, invalid
|
||||
}
|
||||
}
|
||||
if json.Unmarshal(raw, &c) != nil || c.Version != 1 || c.Namespace != ns || c.Filter != filter || c.ID.Validate() != nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 {
|
||||
return c, invalid
|
||||
}
|
||||
if in.CreatedFrom != nil && c.Created.Before(*in.CreatedFrom) || in.CreatedBefore != nil && !c.Created.Before(*in.CreatedBefore) {
|
||||
return c, invalid
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// 所有可选组合均使用固定索引名，不拼接用户输入；队列筛选在LIMIT之前完成。
|
||||
func workflowListSQL(ns core.ID, in instance.ListInput, queues []string, filter string) (string, []any, error) {
|
||||
index := "ix_workflow"
|
||||
if in.DefinitionID != "" {
|
||||
index += "_definition"
|
||||
}
|
||||
if in.BusinessKey != "" {
|
||||
index += "_business"
|
||||
}
|
||||
if in.State != "" {
|
||||
index += "_state"
|
||||
}
|
||||
index += "_created"
|
||||
if in.DefinitionID == "" && in.BusinessKey != "" && in.State == "" {
|
||||
index = "ix_workflow_business"
|
||||
}
|
||||
q := `SELECT i.id,i.namespace_id,i.definition_id,i.definition_version_id,COALESCE(v.version,''),i.state,i.revision,i.paused,i.control_revision,COALESCE(i.business_key,''),i.deadline_at,i.cancel_requested_at,i.stop_kind,i.stop_requested_at,i.completed_at,i.created_at,i.updated_at,
|
||||
i.policy_snapshot,s.queue_ids,COALESCE(s.definition_id=i.definition_id AND s.content_digest=v.content_digest AND v.definition_id=i.definition_id AND d.id IS NOT NULL AND g.id IS NOT NULL,FALSE)
|
||||
FROM weave_instances i FORCE INDEX (` + index + `)
|
||||
LEFT JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.id=i.definition_version_id
|
||||
LEFT JOIN weave_version_scopes s ON s.namespace_id=i.namespace_id AND s.definition_version_id=i.definition_version_id
|
||||
LEFT JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id
|
||||
LEFT JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id
|
||||
WHERE i.namespace_id=?`
|
||||
args := []any{ns}
|
||||
if in.DefinitionID != "" {
|
||||
q += " AND i.definition_id=?"
|
||||
args = append(args, in.DefinitionID)
|
||||
}
|
||||
if in.BusinessKey != "" {
|
||||
q += " AND i.business_key=?"
|
||||
args = append(args, in.BusinessKey)
|
||||
}
|
||||
if in.State != "" {
|
||||
q += " AND i.state=?"
|
||||
args = append(args, in.State)
|
||||
}
|
||||
if in.Version != "" {
|
||||
q += " AND v.version=?"
|
||||
args = append(args, in.Version)
|
||||
}
|
||||
if in.Paused != nil {
|
||||
q += " AND i.paused=?"
|
||||
args = append(args, *in.Paused)
|
||||
}
|
||||
if in.CreatedFrom != nil {
|
||||
q += " AND i.created_at>=?"
|
||||
args = append(args, *in.CreatedFrom)
|
||||
}
|
||||
if in.CreatedBefore != nil {
|
||||
q += " AND i.created_at<?"
|
||||
args = append(args, *in.CreatedBefore)
|
||||
}
|
||||
if in.Cursor != "" {
|
||||
c, err := decodeWorkflowCursor(in.Cursor, ns, in, filter)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
q += " AND (i.created_at<? OR (i.created_at=? AND i.id<?))"
|
||||
args = append(args, c.Created, c.Created, c.ID)
|
||||
}
|
||||
if !slices.Contains(queues, "*") {
|
||||
allowed, err := json.Marshal(queues)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
// 缺失或错配的索引不当作空权限；进入下方完整校验并使整页失败。
|
||||
q += ` AND (s.definition_version_id IS NULL OR v.id IS NULL OR s.definition_id<>i.definition_id OR s.content_digest<>v.content_digest OR JSON_CONTAINS(?,s.queue_ids))`
|
||||
args = append(args, string(allowed))
|
||||
}
|
||||
q += " ORDER BY i.created_at DESC,i.id DESC LIMIT ?"
|
||||
args = append(args, in.Limit+1)
|
||||
return q, args, nil
|
||||
}
|
||||
|
||||
type workflowListRow struct {
|
||||
view instance.Summary
|
||||
policy, queues []byte
|
||||
related bool
|
||||
}
|
||||
|
||||
func (s *WorkflowInstanceStore) List(ctx context.Context, t auth.Token, in instance.ListInput) (instance.Page, error) {
|
||||
out := instance.Page{Items: []instance.Summary{}}
|
||||
err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
queues := workflowQueryQueues(p)
|
||||
filter, err := workflowFilterDigest(in, queues)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q, args, err := workflowListSQL(t.NamespaceID, in, queues, filter)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
candidates := make([]workflowListRow, 0, in.Limit+1)
|
||||
for rows.Next() {
|
||||
var r workflowListRow
|
||||
v := &r.view
|
||||
if err = rows.Scan(&v.ID, &v.NamespaceID, &v.DefinitionID, &v.DefinitionVersionID, &v.Version, &v.State, &v.Revision, &v.Paused, &v.ControlRevision, &v.BusinessKey, &v.DeadlineAt, &v.CancelRequestedAt, &v.StopKind, &v.StopRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &r.policy, &r.queues, &r.related); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
candidates = append(candidates, r)
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// 最多limit+1个候选，按固定版本缓存严格原文验证，不读取每个实例的大输入输出。
|
||||
versions := map[core.ID]definition.VersionView{}
|
||||
allQueues := []core.ID{}
|
||||
for _, r := range candidates {
|
||||
if !r.related {
|
||||
return errors.New("workflow query relation or scope mismatch")
|
||||
}
|
||||
v := r.view
|
||||
ver, ok := versions[v.DefinitionVersionID]
|
||||
if !ok {
|
||||
ver, err = readWorkflowVersion(ctx, tx, t.NamespaceID, v.DefinitionID, v.Version)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
versions[v.DefinitionVersionID] = ver
|
||||
allQueues = append(allQueues, ver.Analysis.Queues...)
|
||||
}
|
||||
if ver.ID != v.DefinitionVersionID || ver.DefinitionID != v.DefinitionID || ver.Version != v.Version {
|
||||
return errors.New("workflow query fixed version mismatch")
|
||||
}
|
||||
var scope []core.ID
|
||||
if json.Unmarshal(r.queues, &scope) != nil || scope == nil {
|
||||
return errors.New("workflow query scope invalid")
|
||||
}
|
||||
slices.Sort(scope)
|
||||
if !slices.Equal(scope, ver.Analysis.Queues) {
|
||||
return errors.New("workflow query scope does not match definition")
|
||||
}
|
||||
for _, id := range scope {
|
||||
if !p.AllowsQueue(id) {
|
||||
return fault.NotFound
|
||||
}
|
||||
}
|
||||
var policy workflowPolicy
|
||||
if json.Unmarshal(r.policy, &policy) != nil || policy != (workflowPolicy{1, ver.ID, ver.ContentDigest, ver.Specification.FailurePolicy, ver.Analysis.MaxTaskAttempts}) {
|
||||
return errors.New("workflow frozen policy mismatch")
|
||||
}
|
||||
}
|
||||
// 在任何实例锁之前，整页涉及的队列统一排序；只读列表不锁实例，也不再次读取变化中的状态。
|
||||
slices.Sort(allQueues)
|
||||
if err = authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, slices.Compact(allQueues)); err != nil {
|
||||
return err
|
||||
}
|
||||
more := len(candidates) > in.Limit
|
||||
if more {
|
||||
candidates = candidates[:in.Limit]
|
||||
}
|
||||
for _, r := range candidates {
|
||||
out.Items = append(out.Items, r.view)
|
||||
}
|
||||
if more {
|
||||
last := out.Items[len(out.Items)-1]
|
||||
raw, err := json.Marshal(workflowCursor{1, t.NamespaceID, last.CreatedAt, last.ID, filter})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.NextCursor = base64.RawURLEncoding.EncodeToString(raw)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return instance.Page{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
Reference in New Issue
Block a user