sparrow/services/knowoapi/services/rule_chain.go

package services

import (
	"encoding/json"
	"github.com/gogf/gf/util/guid"
	"sparrow/pkg/models"
	"sparrow/pkg/rpcs"
	"sparrow/pkg/server"
	"sparrow/services/knowoapi/model"
)

// RuleChainService 业务接口
type RuleChainService interface {
	Create(*models.RuleChain) error
	Delete(*models.RuleChain) error
	Update(*models.RuleChainParams) error
	Query(int, int, string, string) ([]models.RuleChain, int, error)
	Get(vendorId, recordId string) (models.RuleChain, error)
	UpdateChainRoot(*models.ChangeRootParams) error
}
type ruleChainService struct {
	model *model.All
}

// NewRuleChainService  new ruleChain service
func NewRuleChainService(model *model.All) RuleChainService {
	return ruleChainService{
		model: model,
	}
}

func (a ruleChainService) Query(pi, ps int, name, vendorId string) ([]models.RuleChain, int, error) {
	return a.model.RuleChain.Query(pi, ps, name, vendorId, false)
}

func (a ruleChainService) Get(vendorId, recordId string) (models.RuleChain, error) {
	return a.model.RuleChain.Get(vendorId, recordId)
}

func (a ruleChainService) Create(ruleChain *models.RuleChain) error {
	ruleChain.RecordId = guid.S()
	if ruleChain.Root {
		result, total, err := a.model.RuleChain.Query(1, 1, "", ruleChain.VendorID, true)
		if err != nil {
			return err
		}
		if total > 0 {
			_, err = a.model.RuleChain.UpdateChainRoot(result[0].RecordId, false)
			if err != nil {
				return err
			}
		}
	}
	err := a.model.RuleChain.Create(ruleChain)
	if err != nil {
		return err
	}
	params := rpcs.ArgsRuleChainAct{
		RuleChainId: ruleChain.RecordId,
		VendorId:    ruleChain.VendorID,
	}
	reply := new(rpcs.ReplyEmptyResult)
	err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.CreateRuleChain", &params, &reply)
	if err != nil {
		server.Log.Errorf("create ruleChan error : %v", err)
	}
	return nil
}

func (a ruleChainService) create(params *models.RuleChainParams) (firstNodeId string, err error) {
	nodeMap := make(map[string]models.RuleNode)
	var inputNodeId string

	for _, v := range params.Cell {
		if v.GetString("shape") != "edge" {
			if v.GetString("shape") == "input-node" {
				inputNodeId = v.GetString("id")
				continue
			}

			ruleNode := models.RuleNode{
				RecordId:      guid.S(),
				RuleChainID:   params.RecordId,
				Configuration: v.GetString("data"),
				Type:          v.GetString("data.type"),
				Name:          v.GetString("data.name"),
				DebugModel:    true,
				Intro:         v.GetString("data.desc"),
			}
			//if v.Shape != "edge" {
			//	if v.Shape == "input-node" {
			//		inputNodeId = v.Source.Cell
			//		continue
			//	}
			//	ruleNode := models.RuleNode{
			//		RecordId:    guid.S(),
			//		RuleChainID: params.RecordId,
			//		Type:        nodeType[v.Data.Type],
			//		Name:        v.Data.Name,
			//		DebugModel:  true,
			//		Intro:       v.Data.Desc,
			//	}

			//switch ruleNode.Type {
			//case "MsgTypeSwitchNode":
			//	ruleNode.Name = "消息类型路由器"
			//case "MsgTypeFilter":
			//	conMap := map[string]interface{}{
			//		"msg_type": v.Data.MesType,
			//	}
			//	marshal, _ := json.Marshal(conMap)
			//	ruleNode.Configuration = fmt.Sprintf("%s", marshal)
			//
			//case "FilterJavascriptNode":
			//	conMap := map[string]string{
			//		"func_body": v.Data.FuncBody,
			//	}
			//	marshal, _ := json.Marshal(conMap)
			//	ruleNode.Configuration = fmt.Sprintf("%s", marshal)
			//case "RestApiRequestNode":
			//	configuration := models.NodeConfiguration{
			//		Url:       v.Data.Url,
			//		Method:    v.Data.Method,
			//		Retry:     v.Data.Retry,
			//		TimeOut:   v.Data.TimeOut,
			//		RetryWait: v.Data.RetryWait,
			//		Headers:   make(map[string]interface{}),
			//	}
			//	if len(v.Data.Headers) > 0 {
			//		for _, header := range v.Data.Headers {
			//			configuration.Headers[header.Key] = header.Value
			//		}
			//	}
			//}

			//str, err := gjson.DecodeToJson(v)
			//if err != nil {
			//	return "", err
			//}

			nodeMap[v.GetString("id")] = ruleNode

			err = a.model.RuleNode.Create(&ruleNode)
			if err != nil {
				return firstNodeId, err
			}
		}
	}

	for _, v := range params.Cell {
		if v.GetString("shape") == "edge" {
			if v.GetString("source.cell") == inputNodeId {
				firstNodeId = nodeMap[v.GetString("target.cell")].RecordId
				continue
			}
			err = a.model.Relation.Create(&models.Relation{
				RecordId: guid.S(),
				//RuleChainId: params.RecordId,
				FromID: nodeMap[v.GetString("source.cell")].RecordId,
				//FromType:     nodeMap[v.Source.Cell].Type,
				ToID:         nodeMap[v.GetString("target.cell")].RecordId,
				ToType:       nodeMap[v.GetString("target.cell")].Type,
				RelationType: v.GetString("data.label"),
			})
			if err != nil {
				return firstNodeId, err
			}
		}
	}

	return firstNodeId, err
}

func (a ruleChainService) Delete(ruleChain *models.RuleChain) error {
	err := a.delete(ruleChain)
	if err != nil {
		return err
	}
	err = a.model.RuleChain.Delete(ruleChain)
	if err != nil {
		return err
	}
	params := rpcs.ArgsRuleChainAct{
		RuleChainId: ruleChain.RecordId,
		VendorId:    ruleChain.VendorID,
	}
	reply := new(rpcs.ReplyEmptyResult)
	err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.DeleteRuleChain", &params, &reply)
	if err != nil {
		server.Log.Errorf("delete ruleChan error : %v", err)
	}
	return nil
}

func (a ruleChainService) delete(ruleChain *models.RuleChain) error {
	node, _, err := a.model.RuleNode.Query(1, 100, ruleChain.RecordId)
	if err != nil {
		return err
	}
	if len(node) > 0 {
		for _, v := range node {
			_ = a.model.RuleNode.Delete(&v)
		}
	}

	//relation, _, err := a.model.Relation.Query(1, 100, ruleChain.RecordId)
	//if err != nil {
	//	return err
	//}
	//if len(relation) > 0 {
	//	for _, v := range relation {
	//		_ = a.model.Relation.DeleteByChainId(&v)
	//	}
	//}
	return nil
}

func (a ruleChainService) Update(params *models.RuleChainParams) error {
	ruleChain, err := a.Get(params.VendorId, params.RecordId)
	if err != nil {
		return err
	}
	err = a.delete(&ruleChain)
	if err != nil {
		return err
	}
	if params.Cell != nil {
		str, err := json.Marshal(params.Cell)
		if err != nil {
			return err
		}
		ruleChain.Configuration = string(str)
		ruleChain.FirstRuleNodeID, err = a.create(params)
		if err != nil {
			return err
		}
	}

	err = a.model.RuleChain.Update(&ruleChain)
	if err != nil {
		return err
	}
	args := rpcs.ArgsRuleChainAct{
		RuleChainId: ruleChain.RecordId,
		VendorId:    ruleChain.VendorID,
	}
	reply := new(rpcs.ReplyEmptyResult)
	err = server.RPCCallByName(nil, rpcs.ControllerName, "Controller.UpdateRuleChain", &args, &reply)
	if err != nil {
		server.Log.Errorf("update ruleChan error : %v", err)
	}
	return nil
}

func (a ruleChainService) UpdateChainRoot(params *models.ChangeRootParams) error {
	//var isRoot int = 0
	//if params.Root {
	//	isRoot = 1
	//}
	_, err := a.model.RuleChain.UpdateChainRoot(params.RecordId, params.Root)
	if err != nil {
		return err
	}
	if params.Root {
		data, total, err := a.model.RuleChain.Query(1, 1, "", params.VendorId, true)
		if err != nil {
			return err
		}
		if total > 0 {
			data[0].Root = false
			_, err := a.model.RuleChain.UpdateChainRoot(data[0].RecordId, data[0].Root)
			if err != nil {
				return err
			}
		}
	}
	return nil
}