package job

import (
	"context"
	_ "embed"
	"encoding/json"
	"fmt"
	"io"
	"maps"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/juju/errors"
	log "github.com/sirupsen/logrus"
	batchapi "k8s.io/api/batch/v1"
	apiv1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"sigs.k8s.io/yaml"

	"github.com/tidbcloud/auto-deploy/internal/constants"
	"github.com/tidbcloud/auto-deploy/internal/types"
	"github.com/tidbcloud/auto-deploy/internal/utils"
	"github.com/tidbcloud/auto-deploy/internal/utils/s3"
)

const (
	DOCKER_DEFAULT_USER_HOME = "/root"
)

//go:embed job_input.yml.tpl
var JOB_INPUT_TEMPLATE string

type JobState int

type Nothing struct{}

const (
	JOB_STATE_WAITING JobState = iota
	JOB_STATE_RUNNING
	JOB_STATE_SKIPPED
	JOB_STATE_SUCCEEDED
	JOB_STATE_FAILED
	JOB_STATE_DEPENDED_JOB_FAILED
)

type Job struct {
	BatchJob      *batchapi.Job
	State         JobState       // job state
	Version       string         // job version
	Finalizer     string         // finalizer job
	AssociatedJob string         // associated deploy job, only for finalizer job
	JobPath       string         // job definition path
	InputPath     string         // input s3 path
	OutputPath    string         // output s3 path
	Error         error          // if job failed
	Depends       []string       // depended jobs
	CreateAt      *time.Time     // create time
	EndAt         *time.Time     // end time
	Duration      *time.Duration // job execution duration
}

type JobRunner struct {
	region         *types.Region
	clientset      *kubernetes.Clientset
	commonJobInput map[string]any
	jobs           map[string]*Job
	finalizers     map[string]*Job
}

func parseJob(jobPath string) (*Job, error) {
	log.WithField("path", jobPath).Debug("Parse job file")
	batchJob := batchapi.Job{}
	yamlContent, err := os.ReadFile(jobPath)
	if err != nil {
		return nil, errors.Annotatef(err, "Failed to read job file: %s", jobPath)
	}
	jsonContent, err := yaml.YAMLToJSON(yamlContent)
	if err != nil {
		return nil, errors.Annotatef(err, "Failed to convert job yaml to json: %s", jobPath)
	}
	if err := json.Unmarshal(jsonContent, &batchJob); err != nil {
		return nil, errors.Annotatef(err, "Failed to unmarshal job: %s", jobPath)
	}
	job := Job{
		BatchJob:  &batchJob,
		State:     JOB_STATE_WAITING,
		Version:   batchJob.ObjectMeta.Annotations[constants.JOB_VERSION_ANNOTATION],
		Finalizer: batchJob.ObjectMeta.Annotations[constants.JOB_FINALIZER_ANNOTATION],
		JobPath:   jobPath,
		Depends:   GetJobDepends(&batchJob),
	}
	log.WithFields(log.Fields{
		"path":       jobPath,
		"name":       job.BatchJob.Name,
		"version":    job.Version,
		"finalizer":  job.Finalizer,
		"depends-on": job.Depends,
	}).Debug("Parsed job definition")
	return &job, nil
}

func GetJobDepends(job *batchapi.Job) []string {
	if job.ObjectMeta.Annotations[constants.JOB_DEPENDS_ON_ANNOTATION] == "" {
		return []string{}
	}
	depends := strings.Split(job.ObjectMeta.Annotations[constants.JOB_DEPENDS_ON_ANNOTATION], ",")
	depends = utils.Map(depends, func(s string) string { return strings.Trim(s, " ") })
	return depends
}

func NewJobRunner(
	kubeconfig string,
	region *types.Region,
	previousOutputs map[string]string,
) (*JobRunner, error) {
	config, err := clientcmd.RESTConfigFromKubeConfig([]byte(kubeconfig))
	if err != nil {
		return nil, errors.Annotatef(err, "Failed to load kubeconfig: %s", kubeconfig)
	}
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, errors.Annotatef(err, "Failed to create kuberneets client set: %s", kubeconfig)
	}

	err = ApplyCloudAuthSecret(clientset, region.K8s.CredentialSecret)
	if err != nil {
		return nil, err
	}
	commonSecret := utils.Merge(region.OidcProvider.GetEnvMap(), map[string]string{
		constants.GH_TOKEN_DOCKER_ENV: os.Getenv(constants.GH_TOKEN_ENV),
	})
	err = ApplyCommonSecret(clientset, commonSecret)
	if err != nil {
		return nil, err
	}
	err = ApplyDeployJobRunnerServiceAccount(clientset)
	if err != nil {
		return nil, err
	}
	err = ApplyDeployJobRunnerServiceAccountRoleBinding(clientset)
	if err != nil {
		return nil, err
	}

	finalizersPath := region.JobPath + "/finalizers"
	finalizers := make(map[string]*Job)
	entries, err := os.ReadDir(finalizersPath)
	if err != nil && !os.IsNotExist(err) { // ignore no finalizers
		return nil, errors.Annotatef(err, "Failed to read job finalizer folder: %s", region.JobPath)
	}
	log.WithField("count", len(entries)).Debug("Get finalizers")

	for _, e := range entries {
		// Only yaml file
		if !strings.HasSuffix(e.Name(), ".yaml") && !strings.HasSuffix(e.Name(), ".yml") {
			continue
		}

		finalizer, err := parseJob(filepath.Join(finalizersPath, e.Name()))
		if err != nil {
			return nil, err
		}
		if finalizer.Finalizer != "" {
			return nil, errors.NotValidf("Finalizer job %s finalizer", finalizer.BatchJob.Name)
		}
		if len(finalizer.Depends) > 0 {
			return nil, errors.NotValidf("Finalizer job %s dependon job", finalizer.BatchJob.Name)
		}

		finalizer.InputPath = getJobOutputPath(region, fmt.Sprintf("finalizer/%s/%s-input.yaml", finalizer.BatchJob.Name, region.LatestRunId))
		finalizer.OutputPath = getJobOutputPath(region, fmt.Sprintf("finalizer/%s/%s-output.yaml", finalizer.BatchJob.Name, region.LatestRunId))
		MoutCloudAuthSecret(finalizer.BatchJob, region.K8s.CredentialSecretName,
			region.CloudProvider.GetAuthConfigPath(DOCKER_DEFAULT_USER_HOME),
			region.CloudProvider.GetAuthConfigSubPath()) // default user is root, home dir is /root
		AppendJobEnvs(finalizer.BatchJob, finalizer.InputPath, commonSecret)
		SetImagePrefix(finalizer.BatchJob, *region)
		finalizers[finalizer.BatchJob.Name] = finalizer
	}

	jobs := make(map[string]*Job)
	entries, err = os.ReadDir(region.JobPath)
	if err != nil {
		return nil, errors.Annotatef(err, "Failed to read job folder: %s", region.JobPath)
	}
	log.WithField("count", len(entries)).Debug("Get jobs")

	for _, e := range entries {
		// Only yaml file
		if !strings.HasSuffix(e.Name(), ".yaml") && !strings.HasSuffix(e.Name(), ".yml") {
			continue
		}

		job, err := parseJob(filepath.Join(region.JobPath, e.Name()))
		if err != nil {
			return nil, err
		}
		job.InputPath = getJobOutputPath(region, fmt.Sprintf("%s/%s-input.yaml", job.BatchJob.Name, region.LatestRunId))
		job.OutputPath = getJobOutputPath(region, fmt.Sprintf("%s/%s-output.yaml", job.BatchJob.Name, region.LatestRunId))
		MoutCloudAuthSecret(job.BatchJob, region.K8s.CredentialSecretName,
			region.CloudProvider.GetAuthConfigPath(DOCKER_DEFAULT_USER_HOME),
			region.CloudProvider.GetAuthConfigSubPath()) // default user is root, home dir is /root
		AppendJobEnvs(job.BatchJob, job.InputPath, commonSecret)
		SetImagePrefix(job.BatchJob, *region)
		jobs[job.BatchJob.Name] = job
	}
	if err = FlatJobDepend(jobs); err != nil {
		return nil, err
	}
	if err = GetFinalizerJobDepend(jobs, finalizers); err != nil {
		return nil, err
	}

	r := &JobRunner{
		region:         region,
		clientset:      clientset,
		jobs:           jobs,
		finalizers:     finalizers,
		commonJobInput: make(map[string]any),
	}
	// job has succeed in previous run
	for jobName, output := range previousOutputs {
		job, ok := r.jobs[jobName]
		if !ok {
			return nil, errors.NotFoundf("Job %s", jobName)
		}
		job.OutputPath = output
		job.State = JOB_STATE_SKIPPED
	}
	r.commonJobInput, err = utils.RenderJobOptionsTemplate(JOB_INPUT_TEMPLATE, region)
	if err == nil {
		r.commonJobInput[constants.JOB_TEMPLATE_JOB_OPTIONS] = region.JobOptions
	}
	return r, err
}

func getJobOutputPath(region *types.Region, path string) string {
	switch region.CloudProviderType {
	case types.AWS:
		return fmt.Sprintf("s3://%s/%s", region.OutputBucket, path)
	case types.Azure:
		return fmt.Sprintf("%s/%s", region.OutputBucket, path)
	case types.GCP:
		return fmt.Sprintf("gs://%s/%s", region.OutputBucket, path)
	case types.AliCloud:
		return fmt.Sprintf("oss://%s/%s", region.OutputBucket, path)
	default:
		panic("Not supported")
	}
}

func FlatJobDepend(jobs map[string]*Job) error {
	for _, job := range jobs {
		if len(job.Depends) > 0 {
			if err := GetDependedJobs(jobs, job); err != nil {
				return err
			}
		}
	}
	return nil
}

func GetFinalizerJobDepend(jobs, finalizers map[string]*Job) error {
	for jobName, job := range jobs {
		if job.Finalizer != "" {
			finalizer, ok := finalizers[job.Finalizer]
			if !ok {
				return errors.NotFoundf("Job %s finalizer %s", job.BatchJob.Name, job.Finalizer)
			}

			if finalizer.AssociatedJob != "" {
				return errors.NotValidf("Two jobs (%s, %s) have same finalizer %s", finalizer.AssociatedJob, jobName, job.Finalizer)
			}
			finalizer.AssociatedJob = jobName
		}
	}

	for finalizerName, finalizer := range finalizers {
		if finalizer.AssociatedJob == "" {
			return errors.NotFoundf("Finalizer job %s associated job", finalizerName)
		}
		job := jobs[finalizer.AssociatedJob]
		for _, dep := range job.Depends {
			dependJob, ok := jobs[dep]
			if !ok { // should be OK, we have checked all depended jobs
				return errors.NotFoundf("Depended job %s", dep)
			}
			if dependJob.Finalizer != "" {
				dependingFinalizer, ok := finalizers[dependJob.Finalizer]
				if !ok { // should be OK, we have checked all finalizers
					return errors.NotFoundf("Depending finalizer job %s", dependingFinalizer)
				}
				dependingFinalizer.Depends = append(dependingFinalizer.Depends, finalizerName)
			}
		}
	}

	return nil
}

func GetDependedJobs(jobs map[string]*Job, job *Job) error {
	if len(job.Depends) == 0 {
		return nil
	}
	dependJobSet := make(map[string]Nothing)
	for _, jobName := range job.Depends {
		dependJobSet[jobName] = Nothing{}
	}

	if _, ok := dependJobSet[job.BatchJob.Name]; ok {
		return errors.NotValidf("Job %s depends itself", job.BatchJob.Name)
	}

	newAddedJobNames := slices.Clone(job.Depends)
	for len(newAddedJobNames) > 0 {
		tempJobNames := make([]string, 0, 10)
		tempJobNames, newAddedJobNames = newAddedJobNames, tempJobNames // swap temp
		for _, jobName := range tempJobNames {
			newJob, ok := jobs[jobName]
			if !ok {
				return errors.NotFoundf("Depended job %s", jobName)
			}

			for _, newJobName := range newJob.Depends {
				if newJobName == job.BatchJob.Name {
					return errors.NotValidf("Circle job %s depends", job.BatchJob.Name)
				}
				// has added
				if _, ok := dependJobSet[newJobName]; ok {
					continue
				}
				dependJobSet[newJobName] = Nothing{}
				newAddedJobNames = append(newAddedJobNames, newJobName)
			}
		}
	}

	if len(dependJobSet) > len(job.Depends) {
		oldDepends := job.Depends
		job.Depends = make([]string, 0, len(dependJobSet))
		for jobName, _ := range dependJobSet {
			job.Depends = append(job.Depends, jobName)
		}
		log.WithFields(log.Fields{
			"name": job.BatchJob.Name,
			"old":  oldDepends,
			"new":  job.Depends,
		}).Info("Update job depends")
	}

	return nil
}

func (r *JobRunner) prepareJobInput(job *Job) error {
	dependJobsOutput := make(map[string]string)
	for _, dependJobName := range job.Depends {
		if dependJob, ok := r.jobs[dependJobName]; ok {
			dependJobsOutput[dependJobName] = dependJob.OutputPath
		} else {
			return errors.NotFoundf("Job %s output is not found", dependJobName)
		}
	}

	jobInput := maps.Clone(r.commonJobInput)
	jobInput[constants.JOB_TEMPLATE_OUTPUT_PATH] = job.OutputPath
	jobInput[constants.JOB_TEMPLATE_JOB_OUTPUTS] = dependJobsOutput
	content, err := yaml.Marshal(jobInput)
	if err != nil {
		return errors.Annotatef(err, "Failed to marshal job input")
	}
	log.WithFields(log.Fields{
		"job":   job.BatchJob.Name,
		"input": string(content),
	}).Debug("Generate job input content")
	return r.WriteJobInput(job.InputPath, content)
}

func (r *JobRunner) RunAll(ctx context.Context, runId string) {
	for {
		for _, job := range r.jobs {
			r.runJob(ctx, job)
		}

		time.Sleep(constants.DEFAULT_JOB_CHECK_INTERVAL * time.Second)
		allDone := true
		for _, job := range r.jobs {
			if job.State == JOB_STATE_RUNNING {
				r.UpdateJobState(ctx, job)
				if job.State != JOB_STATE_RUNNING {
					err := GetJobLogs(ctx, r.clientset, job.BatchJob.Name)
					if err != nil {
						log.WithFields(log.Fields{
							"job": job.BatchJob.Name,
							"err": err,
						}).Error("Failed to get log")
					}
				}
			}
			if job.State == JOB_STATE_RUNNING || job.State == JOB_STATE_WAITING {
				log.WithField("job", job.BatchJob.Name).Trace("Have running or waiting job")
				allDone = false
			}
		}

		if allDone {
			break
		}
	}
}

func (r *JobRunner) runJob(ctx context.Context, job *Job) {
	if job.State != JOB_STATE_WAITING {
		return
	}

	canRun := true
	for _, dependJobName := range job.Depends {
		dependJob := r.jobs[dependJobName]
		if dependJob.State == JOB_STATE_WAITING || dependJob.State == JOB_STATE_RUNNING {
			canRun = false
			log.WithFields(log.Fields{
				"job":        job.BatchJob.Name,
				"depend-job": dependJobName,
			}).Trace("Wait to run depended job")
		}

		if dependJob.State == JOB_STATE_FAILED || dependJob.State == JOB_STATE_DEPENDED_JOB_FAILED {
			job.State = JOB_STATE_DEPENDED_JOB_FAILED
			job.Error = dependJob.Error
			canRun = false
			log.WithFields(log.Fields{
				"job":          job.BatchJob.Name,
				"depended-job": dependJobName,
			}).Info("Depended job failed")
			break
		}
	}

	if canRun {
		err := r.prepareJobInput(job)
		if err == nil {
			err = r.createBatchJob(ctx, job)
		}

		if err != nil {
			SetJobFinalState(job, err, nil)
			log.WithFields(log.Fields{
				"job": job.BatchJob.Name,
				"err": err,
			}).Info("Failed to create bactch job")
		} else {
			log.WithFields(log.Fields{
				"job": job.BatchJob.Name,
			}).Debug("Job is running")
			job.State = JOB_STATE_RUNNING
		}
	}
}

func AppendJobEnvs(job *batchapi.Job, configPath string, secrets map[string]string) {
	for key := range secrets {
		AppendSecretEnv(job, constants.COMMON_SECRET, key)
	}
	AppendEnv(job, constants.CONFIG_PATH_DOCKER_ENV, configPath)
	// AppendEnv(job, constants.GH_TOKEN_DOCKER_ENV, os.Getenv(constants.GH_TOKEN_ENV))
	AppendEnv(job, constants.GITHUB_SHA_DOCKER_ENV, os.Getenv(constants.GITHUB_SHA_ENV))
	AppendEnv(job, constants.GITHUB_REF_DOCKER_ENV, os.Getenv(constants.GITHUB_REF_ENV))
	AppendEnv(job, constants.JOB_TIMEOUT_DOCKER_ENV, strconv.Itoa(constants.DEFAULT_JOB_TIMEOUT))
	AppendEnv(job, constants.JOB_CHECK_INTERVAL_DOCKER_ENV, strconv.Itoa(constants.DEFAULT_JOB_CHECK_INTERVAL))
}

func SetImagePrefix(job *batchapi.Job, region types.Region) {
	containers := job.Spec.Template.Spec.Containers
	for i := range containers {
		newImage := region.ImageRegistry + constants.IAMGE_PREFIX + containers[i].Image
		log.WithFields(log.Fields{
			"job":       job.Name,
			"container": containers[i].Name,
			"old-image": containers[i].Image,
			"new-image": newImage,
		}).Debug("Update job image")
		containers[i].Image = newImage
	}
	initContainers := job.Spec.Template.Spec.InitContainers
	for i := range initContainers {
		newImage := region.ImageRegistry + constants.IAMGE_PREFIX + initContainers[i].Image
		log.WithFields(log.Fields{
			"job":       job.Name,
			"container": initContainers[i].Name,
			"old-image": initContainers[i].Image,
			"new-image": newImage,
		}).Debug("Update job image")
		initContainers[i].Image = newImage
	}
}

func AppendEnv(job *batchapi.Job, name, value string) {
	containers := job.Spec.Template.Spec.Containers
	for i := range containers {
		envs := containers[i].Env
		hasSet := false
		for j := range envs {
			if envs[j].Name == name {
				log.WithFields(log.Fields{
					"job":       job.Name,
					"name":      name,
					"old-value": envs[j].Value,
					"new-value": value,
				}).Debug("Update enviroment variable")

				envs[j].Value = value
				hasSet = true
			}
		}
		if !hasSet {
			envs = append(envs, apiv1.EnvVar{
				Name:  name,
				Value: value,
			})
			log.WithFields(log.Fields{
				"job":   job.Name,
				"name":  name,
				"value": value,
			}).Debug("Add enviroment variable")
		}
		containers[i].Env = envs
	}
}

func AppendSecretEnv(job *batchapi.Job, secretName, keyName string) {
	valueFrom := &apiv1.EnvVarSource{
		SecretKeyRef: &apiv1.SecretKeySelector{
			LocalObjectReference: apiv1.LocalObjectReference{
				Name: secretName,
			},
			Key: keyName,
		},
	}

	containers := job.Spec.Template.Spec.Containers
	for i := range containers {
		envs := containers[i].Env
		hasSet := false
		for j := range envs {
			if envs[j].Name == keyName {
				log.WithFields(log.Fields{
					"job":  job.Name,
					"name": keyName,
				}).Debug("Update secret enviroment variable")

				envs[j].Value = ""
				envs[j].ValueFrom = valueFrom
				hasSet = true
			}
		}
		if !hasSet {
			envs = append(envs, apiv1.EnvVar{
				Name:      keyName,
				ValueFrom: valueFrom,
			})
			log.WithFields(log.Fields{
				"job":  job.Name,
				"name": keyName,
			}).Debug("Add secret enviroment variable")
		}
		containers[i].Env = envs
	}
}

func (r *JobRunner) createBatchJob(ctx context.Context, job *Job) error {
	job.CreateAt = utils.Ptr(time.Now())
	log.WithFields(log.Fields{
		"job":     job.BatchJob.Name,
		"version": job.Version,
		"input":   job.InputPath,
		"output":  job.OutputPath,
	}).Info("Run job")

	client := r.clientset.BatchV1().Jobs(apiv1.NamespaceDefault)
	// delete old job if exist
	backgroundDeletion := metav1.DeletePropagationBackground
	_ = client.Delete(ctx, job.BatchJob.Name, metav1.DeleteOptions{
		PropagationPolicy: &backgroundDeletion,
	})
	if _, err := client.Create(ctx, job.BatchJob, metav1.CreateOptions{}); err != nil {
		return errors.Annotatef(err, "Failed to create job: %s", job.BatchJob.Name)
	}
	return nil
}

func (r *JobRunner) UpdateJobState(ctx context.Context, job *Job) {
	client := r.clientset.BatchV1().Jobs(apiv1.NamespaceDefault)
	batchJob, err := client.Get(ctx, job.BatchJob.Name, metav1.GetOptions{})
	if err != nil {
		SetJobFinalState(job, errors.Annotatef(err, "Failed to get job: %s", job.BatchJob.Name), batchJob)
		return
	}

	if len(batchJob.Status.Conditions) == 0 {
		log.WithField("job", job.BatchJob.Name).Debug("Job still be active")
		return
	}

	for _, cond := range batchJob.Status.Conditions {
		if cond.Type == batchapi.JobComplete && cond.Status == apiv1.ConditionTrue {
			SetJobFinalState(job, nil, batchJob)
			log.WithField("job", job.BatchJob.Name).Info("Job is succeeded")
			return
		}
		if cond.Type == batchapi.JobFailed && cond.Status == apiv1.ConditionTrue {
			SetJobFinalState(job, errors.NotYetAvailablef(cond.Reason), batchJob)
			log.WithFields(log.Fields{
				"job":     job.BatchJob.Name,
				"reason":  cond.Reason,
				"message": cond.Message,
			}).Info("Job is failed")
			return
		}
	}
}

func (r *JobRunner) RunFinalizers(ctx context.Context) {
	for {
		for _, finalizer := range r.finalizers {
			r.runFinalizer(ctx, finalizer)
		}

		time.Sleep(constants.DEFAULT_JOB_CHECK_INTERVAL * time.Second)
		allDone := true
		for _, finalizer := range r.finalizers {
			if finalizer.State == JOB_STATE_RUNNING {
				r.UpdateJobState(ctx, finalizer)
				if finalizer.State != JOB_STATE_RUNNING {
					err := GetJobLogs(ctx, r.clientset, finalizer.BatchJob.Name)
					if err != nil {
						log.WithFields(log.Fields{
							"job": finalizer.BatchJob.Name,
							"err": err,
						}).Error("Failed to get log")
					}
				}
			}
			if finalizer.State == JOB_STATE_RUNNING || finalizer.State == JOB_STATE_WAITING {
				log.WithField("job", finalizer.BatchJob.Name).Trace("Have running or waiting job")
				allDone = false
			}
		}

		if allDone {
			break
		}
	}
}

func (r *JobRunner) runFinalizer(ctx context.Context, finalizer *Job) {
	if finalizer.State != JOB_STATE_WAITING {
		return
	}

	canRun := true
	for _, dependFinalizerName := range finalizer.Depends {
		dependFinalizer := r.finalizers[dependFinalizerName]
		if dependFinalizer.State == JOB_STATE_WAITING || dependFinalizer.State == JOB_STATE_RUNNING {
			canRun = false
			log.WithFields(log.Fields{
				"finalizer":        finalizer.BatchJob.Name,
				"depend-finalizer": dependFinalizerName,
			}).Trace("Wait to run depended finalizer")
		}

		if dependFinalizer.State == JOB_STATE_FAILED || dependFinalizer.State == JOB_STATE_DEPENDED_JOB_FAILED {
			finalizer.State = JOB_STATE_DEPENDED_JOB_FAILED
			finalizer.Error = dependFinalizer.Error
			canRun = false
			log.WithFields(log.Fields{
				"finalizer":          finalizer.BatchJob.Name,
				"depended-finalizer": dependFinalizerName,
			}).Info("Depended finalizer failed")
			break
		}
	}

	if canRun {
		err := r.prepareFinalizerJobInput(finalizer)
		if err == nil {
			err = r.createBatchJob(ctx, finalizer)
		}

		if err != nil {
			SetJobFinalState(finalizer, err, nil)
			log.WithFields(log.Fields{
				"finalizer": finalizer.BatchJob.Name,
				"err":       err,
			}).Info("Failed to create bactch job")
		} else {
			log.WithFields(log.Fields{
				"finalizer": finalizer.BatchJob.Name,
			}).Debug("finalizer is running")
			finalizer.State = JOB_STATE_RUNNING
		}
	}
}

func (r *JobRunner) prepareFinalizerJobInput(finalizer *Job) error {
	dependJobsOutput := make(map[string]string)
	job := r.jobs[finalizer.AssociatedJob]
	depends := append(job.Depends, job.BatchJob.Name)
	for _, dependJobName := range depends {
		if dependJob, ok := r.jobs[dependJobName]; ok {
			dependJobsOutput[dependJobName] = dependJob.OutputPath
		}
		// FIXME: if the deploymenent is not done, the job may not run. Do we need to run finalizer job?
		// return errors.NotFoundf("Job %s output is not found", dependJobName)
	}

	jobInput := maps.Clone(r.commonJobInput)
	jobInput[constants.JOB_TEMPLATE_OUTPUT_PATH] = finalizer.OutputPath
	jobInput[constants.JOB_TEMPLATE_JOB_OUTPUTS] = dependJobsOutput
	content, err := yaml.Marshal(jobInput)
	if err != nil {
		return errors.Annotatef(err, "Failed to marshal finalizer input")
	}
	log.WithFields(log.Fields{
		"finalizer": finalizer.BatchJob.Name,
		"input":     string(content),
	}).Debug("Generate finalizer input content")
	return r.WriteJobInput(finalizer.InputPath, content)
}

func (r *JobRunner) WriteJobInput(inputPath string, content []byte) error {
	switch r.region.CloudProviderType {
	case types.AWS:
		return s3.WriteFile(inputPath, content)
	case types.Azure:
		return utils.WriteAzureBlobFile(inputPath, content)
	case types.AliCloud:
		return utils.WriteOssFile(inputPath, content)
	default:
		return errors.NotImplementedf("Not support GCP yet")
	}
}

func SetJobFinalState(job *Job, err error, latestJob *batchapi.Job) {
	if err != nil {
		job.State = JOB_STATE_FAILED
		job.Error = err
	} else {
		job.State = JOB_STATE_SUCCEEDED
	}

	if job.CreateAt != nil {
		job.EndAt = utils.Ptr(time.Now())
		UpodateJobDuration(job, latestJob)
	} else {
		job.Duration = utils.Ptr(time.Second * 0)
	}
}

func UpodateJobDuration(job *Job, latestJob *batchapi.Job) {
	startAt := *job.CreateAt
	endAt := *job.EndAt
	if latestJob != nil {
		if latestJob.Status.StartTime != nil {
			startAt = latestJob.Status.StartTime.Time
		}
		if latestJob.Status.CompletionTime != nil {
			endAt = latestJob.Status.CompletionTime.Time
		}
	}
	job.Duration = utils.Ptr(endAt.Sub(startAt))
}

func GetJobLogs(ctx context.Context, clientset *kubernetes.Clientset, jobName string) error {
	jobClient := clientset.BatchV1().Jobs(apiv1.NamespaceDefault)
	podClient := clientset.CoreV1().Pods(apiv1.NamespaceDefault)
	job, err := jobClient.Get(ctx, jobName, metav1.GetOptions{})
	if err != nil {
		log.WithField("job", jobName).WithField("err", err).Error("Failed to get job")
		return errors.Annotatef(err, "Failed to get job: %s", jobName)
	}

	pods, err := podClient.List(ctx, metav1.ListOptions{
		LabelSelector: fmt.Sprintf("job-name=%s,controller-uid=%s", job.Name, job.UID),
		Limit:         1,
	})
	if err != nil {
		log.WithFields(log.Fields{
			"job":      jobName,
			"job-uuid": job.UID,
			"err":      err,
		}).Error("Failed to get pod")
		return errors.Annotatef(err, "Failed to get job pod: %s", jobName)
	}

	var pod *apiv1.Pod
	if len(pods.Items) > 0 {
		pod = &pods.Items[0]
		log.WithFields(log.Fields{
			"job":      jobName,
			"job-uuid": job.UID,
			"pod":      pod.Name,
		}).Debug("Pod found")
	} else {
		return errors.NotFoundf("Job pod not found: %s", jobName)
	}

	req := podClient.GetLogs(pod.Name, &apiv1.PodLogOptions{
		Follow: true,
	})
	reader, err := req.Stream(ctx)
	if err != nil {
		log.WithFields(log.Fields{
			"job": jobName,
			"pod": pod.Name,
			"err": err,
		}).Error("Failed to get pod logs")
		return errors.Annotatef(err, "Failed to get pod logs stream: %s", pod.Name)
	}

	defer reader.Close()
	buf := make([]byte, 1024)
	prefix := fmt.Sprintf("%s-logs: ", jobName)
	logFunc := utils.LogWithPrefix(log.InfoLevel, prefix)
	for {
		count, err := reader.Read(buf)
		if err != nil {
			if err == io.EOF {
				log.WithFields(log.Fields{
					"job": jobName,
					"pod": pod.Name,
				}).Debug("Finished to get logs")
			} else {
				log.WithFields(log.Fields{
					"job": jobName,
					"pod": pod.Name,
					"err": err,
				}).Error("Read logs error")
			}
			break
		}
		logFunc(string(buf[:count]))
	}
	return nil
}