init: 条件流引擎初始提交

condflow.go

@@ -0,0 +1,210 @@
+package condflow
+
+import (
+	"database/sql"
+	"fmt"
+	"sort"
+	"strconv"
+
+	"git.fsdpf.net/go/contracts"
+	"git.fsdpf.net/go/reflux"
+	"git.fsdpf.net/go/reflux/valuex"
+	"git.fsdpf.net/go/req"
+	"github.com/samber/do/v2"
+
+	"git.fsdpf.net/go/db"
+	sqlite3dialect "git.fsdpf.net/go/db/dialect/sqlite3"
+)
+
+type Executor interface {
+	Res() req.Resource
+	// Call(ctx Context, cfg valuex.Accessor) (reflux.R, error)
+}
+
+// Middleware 条件流中间件,用于在执行流程前后扩展处理逻辑
+type Middleware func(ctx Context, cfg valuex.Accessor) error
+
+// CondFlow 条件流服务
+type CondFlow struct {
+	// name 条件流名称
+	name string
+
+	// cases 条件分支列表
+	cases []*CondCase
+
+	// casesDirty 标记条件分支列表是否已变更,用于延迟排序
+	casesDirty bool
+
+	// relations 关联数据, 通过 Orm 查询
+	relations []contracts.Orm
+}
+
+var conn *db.Database
+
+func init() {
+	if dialect, err := sql.Open(sqlite3dialect.DriverWithIF, ":memory:"); err == nil {
+		conn = db.New("sqlite3", dialect)
+	}
+}
+
+// New 创建新的条件流实例
+func New(name string) *CondFlow {
+	return &CondFlow{
+		name:  name,
+		cases: make([]*CondCase, 0),
+	}
+}
+
+// AddCase 添加条件分支
+func (cf *CondFlow) AddCase(c *CondCase) *CondFlow {
+	cf.cases = append(cf.cases, c)
+	cf.casesDirty = true
+	return cf
+}
+
+// Execute 执行条件流
+// ctx: 执行上下文
+// 按照优先级顺序执行匹配的条件分支
+func (cf *CondFlow) Execute(ctx *flowContext) error {
+
+	cases, err := cf.GetValidCondCases(ctx)
+
+	if err != nil {
+		return err
+	}
+
+	for _, ca := range cases {
+		if err := ca.Execute(ctx); err != nil {
+			return err
+		}
+		if !ca.nextCaseOnSuccess {
+			break
+		}
+	}
+
+	return nil
+}
+
+// Run 使用输入数据和用户信息执行条件流
+//
+//   - input 初始输入参数
+//   - user 当前用户信息
+func (cf *CondFlow) Run(app do.Injector, input reflux.R, user req.User) error {
+	ctx := NewFlowContext(app, input, user)
+	return cf.Execute(ctx)
+}
+
+// GetValidCondCases 获取有效的条件分支
+func (cf *CondFlow) GetValidCondCases(ctx *flowContext) ([]*CondCase, error) {
+	input := reflux.New(ctx.Input())
+
+	sd, err := cf.makeTable(input)
+
+	if err != nil {
+		return nil, err
+	}
+
+	cases := cf.getSortedCases()
+
+	cols := []any{}
+	for i, cs := range cases {
+		cols = append(cols, db.Func("IF", cs.cond.ToSql(ctx), 1, 0).As(strconv.Itoa(i)))
+	}
+
+	if len(cols) == 0 {
+		return nil, nil
+	}
+
+	fmt.Println(sd.Select(cols...).Executor().ToSQL())
+
+	result, err := sd.Select(cols...).Executor().GetRecord()
+
+	if err != nil {
+		return nil, err
+	}
+
+	items := []*CondCase{}
+
+	for k, v := range result {
+		if ptr, ok := v.(*any); ok {
+			v = *ptr
+		}
+		val, ok := v.(int32)
+		if !ok {
+			continue
+		}
+		if val == 1 {
+			if n, err := strconv.Atoi(k); err == nil {
+				items = append(items, cases[n])
+			} else {
+				return items, err
+			}
+		}
+	}
+
+	return items, nil
+}
+
+// makeTable 生成条件判断数据表
+func (cf *CondFlow) makeTable(input reflux.R) (*db.SelectDataset, error) {
+	tables := []db.Expression{db.T("Flow"), db.T("Flow").As("@")}
+
+	sd := conn.Select().With("Flow", cf.makeCTE(input))
+
+	for _, re := range cf.relations {
+		result, err := re.Execute(contracts.OrmShow)
+		if err != nil {
+			return nil, err
+		}
+		t := re.GetModel().GetCode()
+		tables = append(tables, db.T(t))
+		sd = sd.With(t, cf.makeCTE(result))
+	}
+
+	return sd.From(tables), nil
+}
+
+// makeCTE 将 reflux.R 数据转换为数据库表达式
+func (cf *CondFlow) makeCTE(src reflux.R) db.Expression {
+	items := []db.Expression{}
+
+	for _, k := range src.Keys() {
+		v := src.Get(k)
+		sqliteType := kindToSQLite3Type(v.Raw().Kind())
+
+		// 根据类型决定使用的值
+		var value any
+		switch sqliteType {
+		case "TEXT":
+			value = v.String()
+		default:
+			value = v.Any()
+		}
+
+		items = append(items, db.Cast(db.V(value), sqliteType).As(k))
+	}
+
+	return db.Select(items)
+}
+
+// getSortedCases 获取按优先级排序的条件分支
+func (cf *CondFlow) getSortedCases() []*CondCase {
+	if cf.casesDirty == true {
+		cf.casesDirty = false
+		sort.Slice(cf.cases, func(i, j int) bool {
+			return cf.cases[i].priority > cf.cases[j].priority
+		})
+	}
+
+	return cf.cases
+}
+
+// Name 获取条件流名称
+func (cf *CondFlow) Name() string {
+	return cf.name
+}
+
+// CaseCount 获取条件分支数量
+func (cf *CondFlow) CaseCount() int {
+	return len(cf.cases)
+}