fix: update task run scheduler migrator

Author: Arta Asadi

services/tasks/db/db.go

@@ -7,7 +7,6 @@ import (
 	"github.com/opengovern/opensecurity/services/tasks/db/models"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
-	"time"
 )
 
 type Database struct {
@@ -136,6 +135,25 @@ func (db Database) FetchLastCreatedTaskRunsByTaskID(taskID string) (*models.Task
 	return &task, nil
 }
 
+// FetchLastTaskRunsByTaskSchedulerID retrieves last task run by scheduler id
+func (db Database) FetchLastTaskRunsByTaskSchedulerID(taskID, taskSchedulerID string) (*models.TaskRun, error) {
+	var task models.TaskRun
+	tx := db.Orm.Model(&models.TaskRun{}).
+		Where("task_id = ?", taskID).
+		Where("trigger_type = ?", models.TriggerTypeScheduled).
+		Where("triggered_by = ?", taskSchedulerID).
+		Order("created_at desc").
+		First(&task)
+	if tx.Error != nil {
+		if errors.Is(tx.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, tx.Error
+	}
+
+	return &task, nil
+}
+
 // TimeoutTaskRunsByTaskID Timeout task runs for given task id by given timeout interval
 func (db Database) TimeoutTaskRunsByTaskID(taskID string, timeoutInterval uint64) error {
 	tx := db.Orm.
@@ -241,11 +259,3 @@ func (db Database) GetTaskConfigSecret(taskId string) (*models.TaskConfigSecret,
 
 	return &configSecret, nil
 }
-
-func (db Database) UpdateTaskRunScheduleLastRun(id uint) error {
-	tx := db.Orm.Model(&models.TaskRunSchedule{}).Where("id = ?", id).Update("last_run", time.Now())
-	if tx.Error != nil {
-		return tx.Error
-	}
-	return nil
-}

services/tasks/db/models/task.go

@@ -4,7 +4,6 @@ import (
 	"github.com/jackc/pgtype"
 	"github.com/lib/pq"
 	"gorm.io/gorm"
-	"time"
 )
 
 type TaskSecretHealthStatus string
@@ -46,9 +45,8 @@ type TaskConfigSecret struct {
 }
 
 type TaskRunSchedule struct {
-	ID        uint `gorm:"primarykey"`
-	TaskID    string
-	LastRun   *time.Time
-	Params    pgtype.JSONB
-	Frequency float64
+	SchedulerID string `gorm:"primarykey"`
+	TaskID      string `gorm:"primarykey"`
+	Params      pgtype.JSONB
+	Frequency   float64
 }

services/tasks/db/models/task_run.go

@@ -17,11 +17,20 @@ const (
 	TaskRunStatusCancelled  TaskRunStatus = "CANCELLED"
 )
 
+type TriggerType string
+
+const (
+	TriggerTypeManual    TriggerType = "MANUAL"
+	TriggerTypeScheduled TriggerType = "SCHEDULED"
+)
+
 type TaskRun struct {
 	gorm.Model
 	TaskID         string
 	Params         pgtype.JSONB
 	Status         TaskRunStatus
 	Result         pgtype.JSONB
+	TriggerType    TriggerType
+	TriggeredBy    string
 	FailureMessage string
 }

services/tasks/scheduler/create_task.go

@@ -40,17 +40,24 @@ func (s *MainScheduler) createTasks(ctx context.Context) error {
 			return err
 		}
 		for _, runSchedule := range runSchedules {
-			if runSchedule.LastRun != nil {
-				if time.Now().Before(runSchedule.LastRun.Add(time.Duration(runSchedule.Frequency) * time.Second)) {
+			lastRun, err := s.db.FetchLastTaskRunsByTaskSchedulerID(task.ID, runSchedule.SchedulerID)
+			if err != nil {
+				return err
+			}
+			if lastRun != nil {
+				if time.Now().Before(lastRun.CreatedAt.Add(time.Duration(runSchedule.Frequency) * time.Second)) {
 					continue
 				}
 			}
+
 			newRun := models.TaskRun{
-				TaskID: task.ID,
-				Status: models.TaskRunStatusCreated,
+				TaskID:      task.ID,
+				Status:      models.TaskRunStatusCreated,
+				TriggerType: models.TriggerTypeScheduled,
+				TriggeredBy: runSchedule.SchedulerID,
 			}
 
-			err := newRun.Result.Set([]byte("{}"))
+			err = newRun.Result.Set([]byte("{}"))
 			if err != nil {
 				return err
 			}
@@ -59,10 +66,6 @@ func (s *MainScheduler) createTasks(ctx context.Context) error {
 			if err = s.db.CreateTaskRun(&newRun); err != nil {
 				return err
 			}
-
-			if err = s.db.UpdateTaskRunScheduleLastRun(runSchedule.ID); err != nil {
-				return err
-			}
 		}
 	}