admin-dashboard-backend/grouptasks/grouptasks.go

package grouptasks

import (
	"encoding/json"
	"fmt"
	"janex/admin-dashboard-backend/modules/cache"
	"janex/admin-dashboard-backend/modules/database"
	"janex/admin-dashboard-backend/modules/structs"
	"janex/admin-dashboard-backend/modules/utils"
	"janex/admin-dashboard-backend/socketclients"
	"os"
	"os/exec"
	"time"

	"github.com/google/uuid"
	"github.com/rs/zerolog/log"
)

var root = "./groupTasks/groups/"

func LoadGroups(category string) {
	entries, err := os.ReadDir(root)

	if err != nil {
		log.Error().Msg("Failed to read groups directory, error: " + err.Error())
		return
	}

	if category != "" {
		cache.RemoveAllCategoryGroupsByCategory(category)
	}

	var updatedGroups []structs.Group

	for _, entry := range entries {
		files, err := os.ReadDir(root + entry.Name())

		if err != nil {
			log.Error().Msg("Failed to read groups directory files, error: " + err.Error())
			return
		}

		for _, file := range files {
			if file.Name() == "index.json" {
				content, err := os.ReadFile(root + entry.Name() + "/index.json")

				if err != nil {
					log.Error().Msg("Failed to read file content, error: " + err.Error())
					return
				}

				var group structs.Group

				json.Unmarshal(content, &group)

				group.Id = entry.Name()

				if category == "" || group.Category == category {
					cache.AddCategoryGroup(group)

					updatedGroups = append(updatedGroups, group)
				}
			}
		}
	}

	if category != "" {
		socketclients.BroadcastMessage(structs.SendSocketMessage{
			Cmd: utils.SentCmdGroupTasksReloaded,
			Body: struct {
				Category      string
				UpdatedGroups []structs.Group
			}{
				Category:      category,
				UpdatedGroups: updatedGroups,
			},
		})
	}
}

const (
	RunGroupTaskStartTypeNormal   = 0
	RunGroupTaskStartTypeTryAgain = 1
)

type RunGroupTaskArgs struct {
	CreatorUserId string
	StartType     uint8
	GroupTaskId   string
	Category      string
	GroupId       string
	Step          uint8
	TaskStepId    string
	GlobalInputs  string
	TaskInputs    string
}

type InputParameters struct {
	ParameterName string `json:"parameterName"`
	Value         string `json:"value"`
}

func RunGroupTask(args RunGroupTaskArgs) {
	categoryGroup := GetCategoryGroupTaskByCategoryAndGroupId(args.Category, args.GroupId)

	groupTaskStep := structs.GroupTaskSteps{
		CreatorUserId: args.CreatorUserId,
		GroupTasksId:  args.GroupTaskId,
		Step:          args.Step,
		Status:        structs.GroupTasksStatusRunning,
		Inputs:        args.TaskInputs,
		StartedAt:     time.Now(),
	}

	// task type
	if args.StartType == RunGroupTaskStartTypeNormal {
		groupTaskStep.Id = uuid.New().String()

		database.DB.Create(&groupTaskStep)

		socketclients.BroadcastMessage(structs.SendSocketMessage{
			Cmd:  utils.SentCmdNewGroupTaskStep,
			Body: groupTaskStep,
		})
	} else if args.StartType == RunGroupTaskStartTypeTryAgain {
		groupTaskStep.Id = args.TaskStepId

		updateGroupTaskSteps(groupTaskStep)
	}

	// set group task to running
	dbGroupTask := updateGroupTask(groupTaskStep.GroupTasksId, structs.GroupTasks{
		Status: structs.GroupTasksStatusRunning,
	})

	// global inputs
	var globalInputParameters []InputParameters

	if len(args.GlobalInputs) > 0 { // global inputs given in args because the group task was just created
		if err := json.Unmarshal([]byte(args.GlobalInputs), &globalInputParameters); err != nil {
			log.Error().Msgf("err unmarshalling global inputs %s", err.Error())
		}
	} else { // global inputs not given in args - fetch it from the database
		if err := json.Unmarshal([]byte(dbGroupTask.GlobalInputs), &globalInputParameters); err != nil {
			log.Error().Msgf("err unmarshalling global inputs %s", err.Error())
		}
	}

	// task parameters
	commandArgs := []string{root + categoryGroup.Id + "/" + categoryGroup.Tasks[args.Step-1].ScriptPath}

	if len(categoryGroup.Tasks[args.Step-1].Parameters) != 0 && len(args.TaskInputs) == 0 {
		updateGroupTask(groupTaskStep.GroupTasksId, structs.GroupTasks{
			Status: structs.GroupTasksStatusInputRequired,
		})

		groupTaskStep.Status = structs.GroupTasksStatusInputRequired

		updateGroupTaskSteps(groupTaskStep)
		return
	} else if len(args.TaskInputs) > 0 {
		var taskParameterInputs []InputParameters

		if err := json.Unmarshal([]byte(args.TaskInputs), &taskParameterInputs); err != nil {
			log.Error().Msgf("err unmarshalling task parameter inputs %s", err.Error())
		}

		for _, taskParameterInput := range taskParameterInputs {
			commandArgs = append(commandArgs, taskParameterInput.Value)
		}
	}

	// execute script
	cmd, err := exec.Command("python3", commandArgs...).Output()

	cmdLog := string(cmd)

	if err != nil {
		if exitErr, ok := err.(*exec.ExitError); ok {
			exitCode := exitErr.ExitCode()
			log.Error().Msgf("exit code %d", exitCode)
			cmdLog += fmt.Sprintf("\nExit code: %v", exitCode)
		}

		log.Error().Msgf("error exec command %s", err.Error())
		groupTaskStep.Status = structs.GroupTasksStatusFailed
	} else {
		groupTaskStep.Status = structs.GroupTasksStatusFinished
	}

	fmt.Println(cmdLog)

	groupTaskStep.Log = cmdLog
	groupTaskStep.EndedAt = time.Now()

	updateGroupTaskSteps(groupTaskStep)

	if int(args.Step) < len(categoryGroup.Tasks) {
		if groupTaskStep.Status == structs.GroupTasksStatusFailed {
			// set group task to failed
			updateGroupTask(groupTaskStep.GroupTasksId, structs.GroupTasks{
				Status: structs.GroupTasksStatusFailed,
			})
		} else {
			args.StartType = RunGroupTaskStartTypeNormal
			args.Step = args.Step + 1

			updateGroupTask(groupTaskStep.GroupTasksId, structs.GroupTasks{
				CurrentTasksStep: args.Step,
			})

			// clear task parameters, because otherwise the next task would have the parameters from the previous task
			args.TaskInputs = ""

			RunGroupTask(args)
		}
	} else {
		// set group task to finished
		updateGroupTask(groupTaskStep.GroupTasksId, structs.GroupTasks{
			Status:  structs.GroupTasksStatusFinished,
			EndedAt: time.Now(),
		})
	}
}

// Updates group task and send it to websocket users
func updateGroupTask(groupTasksId string, updates structs.GroupTasks) structs.GroupTasks {
	database.DB.Model(&structs.GroupTasks{}).Where("id = ?", groupTasksId).Updates(updates)

	var dbGroupTask structs.GroupTasks

	database.DB.First(&dbGroupTask, "id = ?", groupTasksId)

	socketclients.BroadcastMessage(structs.SendSocketMessage{
		Cmd:  utils.SentCmdUpdateGroupTask,
		Body: dbGroupTask,
	})

	return dbGroupTask
}

// Updates group task steps and send it to websocket users
func updateGroupTaskSteps(groupTaskStep structs.GroupTaskSteps) {
	database.DB.Model(&structs.GroupTaskSteps{}).Where("id = ?", groupTaskStep.Id).Updates(groupTaskStep)

	socketclients.BroadcastMessage(structs.SendSocketMessage{
		Cmd:  utils.SentCmdUpdateGroupTaskStep,
		Body: groupTaskStep,
	})
}

func GetAllGroupTasks() []structs.GroupTasks {
	var groupTasks []structs.GroupTasks

	database.DB.Find(&groupTasks)

	return groupTasks
}

func GetAllGroupTasksSteps() []structs.GroupTaskSteps {
	var groupTaskStepsLogs []structs.GroupTaskSteps

	database.DB.Find(&groupTaskStepsLogs)

	return groupTaskStepsLogs
}

func GetCategoryGroupTaskByCategoryAndGroupId(category string, groupId string) structs.Group {
	for _, categoryGroup := range cache.GetCategoryGroups() {
		if categoryGroup.Category == category {
			for _, group := range categoryGroup.Groups {
				if group.Id == groupId {
					return group
				}
			}
		}
	}

	return structs.Group{}
}

func StartUnlockLockedGroupTaskStepsTicker() {
	ticker := time.NewTicker(1 * time.Second)

	for _ = range ticker.C {
		for index, taskStep := range cache.GetLockedGroupTaskSteps() {
			if time.Since(taskStep.LockedAt).Seconds() > 3 {
				log.Debug().Msgf("Unlocked task step", index)
				cache.RemoveLockedGroupTaskStep(index)

				socketclients.BroadcastMessageExceptUserSessionId(taskStep.LockedByUserSession,
					structs.SendSocketMessage{
						Cmd: utils.SentCmdTaskUnlocked,
						Body: struct {
							GroupTaskId string
							Step        uint8
						}{
							GroupTaskId: taskStep.GroupTaskId,
							Step:        taskStep.Step,
						},
					})
			}
		}
	}
}