adm-management/impl/v1/task_management/relation.go

package task_management

import (
	"context"
	"encoding/json"
	"fmt"

	"adm-management/errors"
	"adm-management/model"
	v1 "adm-management/pb/v1"

	"git.getensh.com/common/gopkgsv2/database"
	"git.getensh.com/common/gopkgsv2/logger"
	"go.uber.org/zap"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
)

var a = 1

func inSlice(arr []string, s string) bool {
	for _, v := range arr {
		if v == s {
			return true
		}
	}

	return false
}

// 任务信息 向上的关系
func toTaskId(
	ctx context.Context,
	taskId int64,
	tables []string,
) (reply *v1.DataSourceRelationReply_Node, err error, t []string) {
	// 获取task信息
	task, err := model.NewTaskList().Get(database.DB().Where("task_id=?", taskId))
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, errors.SystemError, t
	}

	if err == gorm.ErrRecordNotFound {
		return reply, nil, t
	}

	if inSlice(tables, task.TaskName) {
		return reply, nil, t
	}

	tables = append(tables, task.TaskName)
	t = append(t, task.TaskName)

	reply = &v1.DataSourceRelationReply_Node{
		NodeName: task.TaskName,
		Desc:     task.Desc,
		IsTask:   true,
	}

	// 获取依赖信息
	dependency, err := model.NewDependency().GetSourceByTaskId(database.DB().Where("t1.task_id = ?", taskId))
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, errors.SystemError, t
	}

	if err == nil {
		if !inSlice(tables, dependency.TableName) {
			reply.Rely = append(reply.Rely, &v1.Rely{
				TableName: dependency.TableName,
				Desc:      dependency.Desc,
			})
			tables = append(tables, dependency.TableName)
			t = append(t, dependency.TableName)
		}
	}

	// 向上
	pagination := model.NewPagination(0, 100, 0).GetLimitOffset()
	relationList, err := model.NewRelation().List(database.DB().Where("task_id = ? and is_on = 1", taskId), pagination)

	for _, v := range relationList {
		source1, err := model.NewDataList().Get(database.DB().Where("source_code = ?", v.SourceCode))
		if err != nil && err != gorm.ErrRecordNotFound {
			return nil, err, t
		}

		entry, err := toSourceCode(ctx, &v1.DataSourceRelationRequest{
			SourceCode: source1.SourceCode,
		}, tables)
		if err != nil {
			return nil, err, t
		}

		if entry != nil {
			reply.Entry = append(reply.Entry, entry)
		}
	}

	return reply, nil, t
}

// 任务信息 只需要向下的关系
func fromTaskId(
	ctx context.Context,
	taskId int64,
	tables []string,
	in *v1.DataSourceRelationReply_Node,
) (reply *v1.DataSourceRelationReply_Node, err error) {
	// 获取task信息
	task, err := model.NewTaskList().Get(database.DB().Where("task_id=?", taskId))
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, errors.SystemError
	}

	if err == gorm.ErrRecordNotFound {
		return reply, nil
	}

	if inSlice(tables, task.TaskName) {
		return reply, nil
	}

	tables = append(tables, task.TaskName)

	if a == 1 {
		reply = in
		a++
	} else {
		reply = &v1.DataSourceRelationReply_Node{
			NodeName: task.TaskName,
			Desc:     task.Desc,
			IsTask:   true,
		}
	}

	// 获取依赖信息
	dependency, err := model.NewDependency().GetSourceByTaskId(database.DB().Where("t1.task_id = ?", taskId))
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, errors.SystemError
	}

	if err == nil {
		if !inSlice(tables, dependency.TableName) {
			reply.Rely = append(reply.Rely, &v1.Rely{
				TableName: dependency.TableName,
				Desc:      dependency.Desc,
			})
			tables = append(tables, dependency.TableName)
		}
	}

	// use TargetTable find TargetSourceCode
	source, err := model.NewDataList().Get(database.DB().Where("table_name = ?", task.TargetTable))
	if err != nil && err != gorm.ErrRecordNotFound {
		return nil, err
	}

	if err == gorm.ErrRecordNotFound {
		reply.Output = []*v1.DataSourceRelationReply_Node{{
			NodeName: task.TargetTable,
			Desc:     source.Desc,
		}}

		return reply, nil
	}

	output, err := fromSourceCode(ctx, &v1.DataSourceRelationRequest{
		SourceCode: source.SourceCode,
	}, tables, nil)
	if err != nil {
		return nil, err
	}

	if output != nil && output.Output != nil {
		reply.Output = []*v1.DataSourceRelationReply_Node{{
			NodeName: task.TargetTable,
			Desc:     source.Desc,
			Output:   output.Output,
		}}
	} else {
		reply.Output = []*v1.DataSourceRelationReply_Node{{
			NodeName: task.TargetTable,
			Desc:     source.Desc,
		}}
	}

	return reply, nil
}

// toSourceCode 向上
func toSourceCode(
	ctx context.Context,
	req *v1.DataSourceRelationRequest,
	tables []string,
) (reply *v1.DataSourceRelationReply_Node, err error) {
	// 获取数据源信息
	source, err := model.NewDataList().Get(database.DB().Where("source_code = ?", req.SourceCode))
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, err
	}

	if err == gorm.ErrRecordNotFound {
		return reply, nil
	}

	if inSlice(tables, source.TableName) {
		return reply, nil
	}

	tables = append(tables, source.TableName)

	reply = &v1.DataSourceRelationReply_Node{
		NodeName: source.TableName,
		Desc:     source.Desc,
	}

	// 向上获取TaskId
	// 构造分页类
	pagination1 := model.NewPagination(0, 100, 0).GetLimitOffset()
	task, err := model.NewTaskList().
		List(database.DB().Where("target_table = ? ", source.TableName), pagination1)
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, errors.SystemError
	}

	if err == gorm.ErrRecordNotFound {
		return reply, nil
	}

	// 获取输入关系
	for _, v := range task {

		node, err, _ := toTaskId(ctx, v.TaskId, tables)
		if err != nil {
			return reply, err
		}

		if node != nil && node.NodeName != "" {
			reply.Entry = append(reply.Entry, node)
		}
	}

	return reply, nil
}

// fromeSourceCode 任务列表需要上下的全部关系
func fromSourceCode(
	ctx context.Context,
	req *v1.DataSourceRelationRequest,
	tables []string,
	in *v1.DataSourceRelationReply_Node,
) (reply *v1.DataSourceRelationReply_Node, err error) {
	// 获取数据源信息
	source, err := model.NewDataList().Get(database.DB().Where("source_code = ?", req.SourceCode))
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, err
	}

	if err == gorm.ErrRecordNotFound {
		return reply, nil
	}

	if inSlice(tables, source.TableName) {
		return reply, nil
	}

	tables = append(tables, source.TableName)

	if a == 1 {
		reply = in
		a++
	} else {
		reply = &v1.DataSourceRelationReply_Node{
			NodeName: source.TableName,
			Desc:     source.Desc,
		}
	}
	// 向下获取TaskId
	// 构造分页类
	pagination := model.NewPagination(0, 100, 0).GetLimitOffset()
	relationList, err := model.NewRelation().
		List(database.DB().Where("source_code = ? AND is_on = 1", req.SourceCode), pagination)
	if err != nil && err != gorm.ErrRecordNotFound {
		return reply, errors.SystemError
	}

	if err == gorm.ErrRecordNotFound {
		return reply, nil
	}

	// 获取输出关系
	for _, v := range relationList {
		node, err := fromTaskId(ctx, v.TaskId, tables, nil)
		if err != nil {
			return reply, err
		}
		if node != nil && node.NodeName != "" {
			reply.Output = append(reply.Output, node)
		}
	}

	return reply, nil
}

// DataSourceRelation 血缘关系
func DataSourceRelation(
	ctx context.Context,
	req *v1.DataSourceRelationRequest,
) (reply *v1.DataSourceRelationReply, err error) {
	reply = &v1.DataSourceRelationReply{}
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("%+v", r)
			e := &status.Status{}
			if er := json.Unmarshal([]byte(err.Error()), e); er != nil {
				logger.Error("err",
					zap.String("system_err", err.Error()),
					zap.Stack("stacktrace"))
			}
		}
	}()

	if req.SourceCode == "" && req.TaskId == 0 {
		return reply, errors.ParamsError
	}

	tables := make([]string, 0, 20)
	a = 1

	// source 需要上下的全部关系
	// task 只需要向下的关系
	switch {
	case req.SourceCode != "":
		reply.Node, err = toSourceCode(ctx, req, tables)
		reply.Node, err = fromSourceCode(ctx, req, tables, reply.Node)
	case req.TaskId != 0:
		reply.Node, err, tables = toTaskId(ctx, req.TaskId, tables)
		tables = tables[1:]
		reply.Node, err = fromTaskId(ctx, req.TaskId, tables, reply.Node)
	}

	if err != nil {
		return reply, err
	}

	return reply, nil
}