condflow/condflow.go

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)
}