Schedule Rotation

Author: hasa1K

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