Merge pull request #2350 from actiontech/fix-issue1441-4

Fix issue1441 4

Author: sjjian

sqle/api/controller/v2/workflow.go

@@ -1021,6 +1021,12 @@ func UpdateWorkflowScheduleV2(c echo.Context) error {
 		return controller.JSONBaseErrorReq(c, err)
 	}
 
+	if req.IsNotify != nil && *req.IsNotify && req.NotifyType != nil {
+		if err := createNotifyRecord(*req.NotifyType, curTaskRecord); err != nil {
+			return controller.JSONBaseErrorReq(c, err)
+		}
+	}
+
 	return c.JSON(http.StatusOK, controller.NewBaseReq(nil))
 }
 

sqle/api/controller/v2/workflow_ce.go

@@ -0,0 +1,11 @@
+//go:build !enterprise
+// +build !enterprise
+
+package v2
+
+import "github.com/actiontech/sqle/sqle/model"
+
+func createNotifyRecord(notifyType string, curTaskRecord *model.WorkflowInstanceRecord) error {
+	// nothing
+	return nil
+}

sqle/model/configuration.go

@@ -285,3 +285,70 @@ func (s *Storage) GetFeishuInstByStatus(status string) ([]FeishuInstance, error)
 	}
 	return feishuInst, nil
 }
+
+type WechatOAStatus int
+
+const (
+	INITIALIZED WechatOAStatus = 1
+	APPROVED    WechatOAStatus = 2
+	REJECTED    WechatOAStatus = 3
+)
+
+type WechatRecord struct {
+	Model
+	TaskId   uint   `json:"task_id" gorm:"column:task_id"`
+	OaResult string `json:"oa_result" gorm:"column:oa_result;default:\"INITIALIZED\""`
+	SpNo     string `json:"sp_no" gorm:"column:sp_no"`
+
+	Task *Task `gorm:"foreignkey:TaskId"`
+}
+
+func (s *Storage) GetWechatRecordByStatus(status string) ([]WechatRecord, error) {
+	var wcRecords []WechatRecord
+	err := s.db.Where("oa_result = ?", status).Find(&wcRecords).Error
+	if err != nil {
+		return nil, err
+	}
+	return wcRecords, nil
+}
+
+func (s *Storage) WechatCancelScheduledTask(w WechatRecord) error {
+	err := s.db.Transaction(func(tx *gorm.DB) error {
+		// 取消该task对应的定时上线任务，将WorkflowInstanceRecord表中的ScheduledAt字段设置为null
+		insRecord := WorkflowInstanceRecord{}
+		err := s.db.Where("task_id = ?", w.TaskId).Find(&insRecord).Error
+		if err != nil {
+			return err
+		}
+		insRecord.ScheduledAt = nil
+		err = s.Save(&insRecord)
+		if err != nil {
+			return err
+		}
+
+		w.OaResult = ApproveStatusRefuse
+		err = s.Save(&w)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	return err
+}
+
+func (s *Storage) GetWechatRecordByTaskId(taskId uint) (*WechatRecord, error) {
+	var wcRecord WechatRecord
+	err := s.db.Where("task_id = ?", taskId).Find(&wcRecord).Error
+	if err != nil {
+		return nil, err
+	}
+	return &wcRecord, nil
+}
+
+func (s *Storage) UpdateWechatRecordByTaskId(taskId uint, m map[string]interface{}) error {
+	err := s.db.Model(&WechatRecord{}).Where("task_id = ?", taskId).Updates(m).Error
+	if err != nil {
+		return errors.New(errors.ConnectStorageError, err)
+	}
+	return nil
+}

sqle/model/utils.go

@@ -149,6 +149,7 @@ var autoMigrateList = []interface{}{
 	&CompanyNotice{},
 	&SqlManageEndpoint{},
 	&SQLDevRecord{},
+	&WechatRecord{},
 }
 
 func (s *Storage) AutoMigrate() error {

sqle/model/workflow.go

@@ -237,13 +237,23 @@ type WorkflowInstanceRecord struct {
 	// 用于区分工单处于上线步骤时，某个数据源是否已上线，因为数据源可以分批上线
 	IsSQLExecuted   bool
 	ExecutionUserId string
+	// 定时上线是否需要发生通知
+	NeedScheduledTaskNotify bool
 
 	Instance *Instance `gorm:"foreignkey:InstanceId"`
 	Task     *Task     `gorm:"foreignkey:TaskId"`
 	// User     *User     `gorm:"foreignkey:ExecutionUserId"`
 	ExecutionAssignees string
 }
 
+func (s *Storage) UpdateWorkflowInstanceRecordById(id uint, m map[string]interface{}) error {
+	err := s.db.Model(&WorkflowInstanceRecord{}).Where("id = ?", id).Updates(m).Error
+	if err != nil {
+		return errors.New(errors.ConnectStorageError, err)
+	}
+	return nil
+}
+
 func (s *Storage) GetWorkInstanceRecordByTaskId(id string) (instanceRecord WorkflowInstanceRecord, err error) {
 	return instanceRecord, s.db.Where("task_id = ?", id).First(&instanceRecord).Error
 }

sqle/pkg/im/im.go

@@ -187,3 +187,55 @@ func BatchCancelApprove(workflowIds []string, user *model.User) {
 		}
 	}
 }
+
+func CreateScheduledApprove(taskId uint, projectId, workflowId string) {
+	newLog := log.NewEntry()
+	s := model.GetStorage()
+
+	workflow, err := dms.GetWorkflowDetailByWorkflowId(projectId, workflowId, s.GetWorkflowDetailWithoutInstancesByWorkflowID)
+	if err != nil {
+		newLog.Errorf("get workflow error: %v", err)
+	}
+	assignUserIds := workflow.CurrentAssigneeUser()
+
+	assignUsers, err := dms.GetUsers(context.TODO(), assignUserIds, controller.GetDMSServerAddress())
+	if err != nil {
+		newLog.Errorf("get user error: %v", err)
+		return
+	}
+
+	ims, err := s.GetAllIMConfig()
+	if err != nil {
+		newLog.Errorf("get im config error: %v", err)
+		return
+	}
+
+	for _, im := range ims {
+		if !im.IsEnable {
+			continue
+		}
+
+		systemVariables, err := s.GetAllSystemVariables()
+		if err != nil {
+			newLog.Errorf("get sqle url system variables error: %v", err)
+			continue
+		}
+
+		sqleUrl := systemVariables[model.SystemVariableSqleUrl].Value
+		workflowUrl := fmt.Sprintf("%v/project/%s/order/%s", sqleUrl, workflow.ProjectId, workflow.WorkflowId)
+		if sqleUrl == "" {
+			newLog.Errorf("sqle url is empty")
+			workflowUrl = ""
+		}
+
+		switch im.Type {
+		case model.ImTypeWechatAudit:
+			if err := CreateWechatAuditRecord(context.TODO(), im, workflow, assignUsers, workflowUrl, taskId); err != nil {
+				newLog.Errorf("create wechat audit error: %v", err)
+				continue
+			}
+		default:
+			newLog.Errorf("im type %s not found", im.Type)
+		}
+	}
+}

sqle/pkg/im/im_ce.go

@@ -49,3 +49,7 @@ func CancelDingdingAuditInst(ctx context.Context, im model.IM, workflowIDs []str
 func CreateWechatAuditTemplate(ctx context.Context, im model.IM) error {
 	return ErrCommunityEditionNotSupportWechatAudit
 }
+
+func CreateWechatAuditRecord(ctx context.Context, im model.IM, workflow *model.Workflow, assignUsers []*model.User, url string, taskId uint) error {
+	return ErrCommunityEditionNotSupportWechatAudit
+}

sqle/server/manager.go

@@ -17,6 +17,7 @@ var OnlyRunOnLeaderJobs = []func(entry *logrus.Entry) ServerJob{
 	NewCleanJob,
 	NewDingTalkJob,
 	NewFeishuJob,
+	NewWechatJob,
 }
 
 var RunOnAllJobs = []func(entry *logrus.Entry) ServerJob{

sqle/server/wechat_ce.go

@@ -0,0 +1,26 @@
+//go:build !enterprise
+// +build !enterprise
+
+package server
+
+import (
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+type WechatJob struct {
+	BaseJob
+}
+
+func NewWechatJob(entry *logrus.Entry) ServerJob {
+	w := new(WechatJob)
+	w.BaseJob = *NewBaseJob(entry, 60*time.Second, w.wechatRotation)
+	return w
+}
+
+// 当前企微轮询只为二次确认工单定时上线功能
+// https://github.com/actiontech/sqle-ee/issues/1441
+func (w *WechatJob) wechatRotation(entry *logrus.Entry) {
+	// nothing
+}

sqle/server/workflow_schedule.go

@@ -15,6 +15,7 @@ import (
 	"github.com/actiontech/sqle/sqle/log"
 	"github.com/actiontech/sqle/sqle/model"
 	"github.com/actiontech/sqle/sqle/notification"
+	"github.com/actiontech/sqle/sqle/pkg/im"
 	"github.com/sirupsen/logrus"
 )
 
@@ -33,13 +34,17 @@ func NewWorkflowScheduleJob(entry *logrus.Entry) ServerJob {
 
 func (j *WorkflowScheduleJob) WorkflowSchedule(entry *logrus.Entry) {
 	st := model.GetStorage()
+
+	entry.Error("start WorkflowSchedule")
 	workflows, err := st.GetNeedScheduledWorkflows()
 	if err != nil {
 		entry.Errorf("get need scheduled workflows from storage error: %v", err)
 		return
 	}
+	entry.Errorf("start WorkflowSchedule workflow:%v", workflows)
 	now := time.Now()
 	for _, workflow := range workflows {
+		entry.Errorf("get workflow id %v", workflow.WorkflowId)
 		w, err := dms.GetWorkflowDetailByWorkflowId(string(workflow.ProjectId), workflow.WorkflowId, st.GetWorkflowDetailWithoutInstancesByWorkflowID)
 		if err != nil {
 			entry.Errorf("get workflow from storage error: %v", err)
@@ -60,6 +65,17 @@ func (j *WorkflowScheduleJob) WorkflowSchedule(entry *logrus.Entry) {
 		needExecuteTaskIds := map[uint]string{}
 		for _, ir := range w.Record.InstanceRecords {
 			if !ir.IsSQLExecuted && ir.ScheduledAt != nil && ir.ScheduledAt.Before(now) {
+				if ir.NeedScheduledTaskNotify {
+					isOaAgree, err := getOaApproveResult(ir.TaskId, workflow)
+
+					if err != nil {
+						entry.Error(err)
+						continue
+					}
+					if !isOaAgree {
+						continue
+					}
+				}
 				needExecuteTaskIds[ir.TaskId] = ir.ScheduleUserId
 			}
 		}
@@ -75,6 +91,21 @@ func (j *WorkflowScheduleJob) WorkflowSchedule(entry *logrus.Entry) {
 	}
 }
 
+func getOaApproveResult(taskId uint, workflow *model.Workflow) (bool, error) {
+	s := model.GetStorage()
+	record, err := s.GetWechatRecordByTaskId(taskId)
+	if err != nil {
+		return false, err
+	}
+	if record.SpNo == "" {
+		go im.CreateScheduledApprove(taskId, string(workflow.ProjectId), workflow.WorkflowId)
+		return false, nil
+	} else if record.OaResult == model.ApproveStatusAgree {
+		return true, nil
+	}
+	return false, nil
+}
+
 func ExecuteWorkflow(workflow *model.Workflow, needExecTaskIdToUserId map[uint]string) error {
 	s := model.GetStorage()