package schedule

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	wfutil "github.com/argoproj/argo-workflows/v3/workflow/util"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow"
	"github.com/pkg/errors"
	"k8s.io/klog/v2"

	. "github.com/pingcap/test-infra/common/model"
	"github.com/pingcap/test-infra/services/tps/pkg/common"
)

type autoRefreshedValue struct {
	value     float64
	updatedAt time.Time
	lock      sync.Mutex
	done      int64
}

func (v *autoRefreshedValue) Get() float64 {
	v.lock.Lock()
	val := v.value
	v.lock.Unlock()
	return val
}

func (v *autoRefreshedValue) Inc(val float64) {
	v.lock.Lock()
	v.value += val
	v.updatedAt = time.Now()
	v.lock.Unlock()
}

func (v *autoRefreshedValue) Stop() {
	atomic.CompareAndSwapInt64(&v.done, 0, 1)
}

const compileErrMsg = "failed to compile plan to workflow"

func submitTestJob(ctx context.Context, job *common.ArgoJob, tcms common.TCMS, argo ArgoClient) error {
	var wf *wfapi.Workflow
	var err error
	if job.Workflow != nil {
		wf, err = argo.Get(ctx, job.Workflow.Name)
		if err != nil {
			klog.ErrorS(err, "fetch argo job fail", "name", job.Workflow.Name)
		}
		if !wf.Status.Phase.Completed() {
			return fmt.Errorf("resubmit uncompleted argo job: %d", job.ID())
		}
		wf.TypeMeta = metav1.TypeMeta{
			Kind:       workflow.WorkflowKind,
			APIVersion: workflow.APIVersion,
		}

		var setupTemplate *wfapi.Template
		for index := range wf.Spec.Templates {
			if role, ok := wf.Spec.Templates[index].Metadata.Labels["role"]; ok {
				if role == "setUp" {
					setupTemplate = &wf.Spec.Templates[index]
					break
				}
			}
		}
		if setupTemplate != nil {
			for index := range setupTemplate.Container.Env {
				if setupTemplate.Container.Env[index].Name == "TESTBED_INPUT_SPEC" {
					var testbed map[string]interface{}
					err = json.Unmarshal([]byte(setupTemplate.Container.Env[index].Value), &testbed)
					if err != nil {
						return err
					}
					if _, ok := testbed["name"]; ok {
						testbed["name"] = job.Plan.TestBed.Name
						input, err := json.Marshal(testbed)
						if err != nil {
							return err
						}
						setupTemplate.Container.Env[index].Value = string(input)
					}
					break
				}
			}
		}

		wf, err = wfutil.FormulateResubmitWorkflow(wf, false)
	} else {
		wf, err = buildArgoWorkflow(ctx, job, tcms)
	}
	if err != nil {
		return err
	}
	wf.Annotations = map[string]string{
		"testbed": job.Plan.TestBed.Name,
	}

	wf, err = argo.Create(ctx, wf)
	if err != nil {
		return err
	}
	name := wf.Name
	if err != nil && !kerrors.IsAlreadyExists(err) {
		return errors.WithStack(err)
	}
	if err != nil {
		klog.InfoS("get existing workflow", "name", name)
		wf, err = argo.Get(ctx, name)
		if err != nil {
			return errors.WithStack(err)
		}
	}
	job.UpdateStatus(wf)
	job.PlanExec.Status = ExecutionStatusPreparing
	if err = tcms.PutPlanExecution(job.PlanExec); err != nil {
		klog.ErrorS(err, "cannot set plan-execution after submitting workflow", "workflow", wf.Name)
	}

	TestBedInfo := TestBedInfo{
		Name:                    job.Plan.TestBed.Name,
		PlanExecutionID:         job.PlanExec.ID,
		URL:                     job.Plan.TestBed.RmsUrl + "/api/v1/testbeds/" + job.Plan.TestBed.Name,
		PhysicalResourceRequest: job.Plan.TestBed.PhysicalResourceRequest,
		ResourcePool: ResourcePool{
			Name: job.Plan.TestBed.ResourcePool,
		},
	}
	if err = tcms.CreateTestBedInfo(&TestBedInfo); err != nil {
		klog.ErrorS(err, "cannot create testbedInfo after submitting workflow", "testbed-name", TestBedInfo.Name)
	}
	return nil
}

func buildArgoWorkflow(ctx context.Context, job *common.ArgoJob, tcms common.TCMS) (*wfapi.Workflow, error) {
	var preparedPlans []TestPlan
	if job.Plan != nil {
		preparedPlans = append(preparedPlans, *job.Plan)
	}
	cc := NewContext(tcms, preparedPlans, job.PlanExec.ID)

	planRef := PlanRef{
		PlanID:        job.PlanExec.PlanID,
		VariableBinds: job.PlanExec.VariableBinds,
	}

	wfSpec, err := CompileArgo(cc, planRef)
	if err != nil {
		return nil, errors.Wrap(err, compileErrMsg)
	}
	name := fmt.Sprintf("plan-exec-%v", job.PlanExec.ID)
	wf := &wfapi.Workflow{
		TypeMeta: metav1.TypeMeta{
			Kind:       workflow.WorkflowKind,
			APIVersion: workflow.APIVersion,
		},
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Spec: *wfSpec,
	}
	return wf, nil
}

// SetErrorTestJob sets the test job as error and reports to TCMS it's finished with the ExecutionStatusError state
func SetErrorTestJob(_ context.Context, job *common.ArgoJob, jobMsg string, tcms common.TCMS) error {
	job.Status = common.ArgoTestJobDone
	job.Message = jobMsg
	job.PlanExec.Status = ExecutionStatusError
	job.PlanExec.Output = appendExecOutput(job.PlanExec.Output, "cannot submit plan: "+job.Message)
	completedAt := Timestamp(time.Now().Unix())
	job.PlanExec.CompletedAt = &completedAt
	return tcms.PutPlanExecution(job.PlanExec)
}

// SetCanceledTestJob sets the plan execution finished with the ExecutionStatusCancelled state
func SetCanceledTestJob(_ context.Context, job *common.ArgoJob, jobMsg string, output string, tcms common.TCMS) error {
	job.Status = common.ArgoTestJobDone
	job.Message = jobMsg
	job.PlanExec.Status = ExecutionStatusCancelled
	job.PlanExec.Output = output

	completedAt := Timestamp(time.Now().Unix())
	job.PlanExec.CompletedAt = &completedAt
	if err := tcms.PutPlanExecution(job.PlanExec); err != nil {
		klog.ErrorS(err, "cannot set plan-execution status to cancelled")
		return err
	}
	return nil
}

func isCompileError(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(err.Error(), compileErrMsg)
}

func appendExecOutput(old string, n string) string {
	if old == "" {
		return n
	}
	return old + "\n" + n
}