package schedule

import (
	"encoding/json"
	"fmt"
	"io"
	"path"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/test-infra/common/model/rms/request"

	wf "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	versionlib "github.com/hashicorp/go-version"
	"github.com/pingcap/test-infra/services/tps/pkg/common"
	"github.com/pkg/errors"
	"github.com/valyala/bytebufferpool"
	"github.com/valyala/fasttemplate"
	corev1 "k8s.io/api/core/v1"
	kres "k8s.io/apimachinery/pkg/api/resource"

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

const (
	testbedOutputGlobalName = "testbed-output"
	teardownTemplateName    = "teardown"
	setupImageParamName     = "setupImage"
	setupImageParamRefName  = "{{workflow.parameters." + setupImageParamName + "}}"
)

func CompileArgo(ctx CompileContext, target PlanRef) (*wf.WorkflowSpec, error) {
	plan, err := compilePlan(ctx, &target, nil)
	if err != nil {
		return nil, err
	}
	return &wf.WorkflowSpec{
		Entrypoint: plan.template.Name,
		OnExit:     teardownTemplateName,
		Templates:  ctx.ListTemplates(),
	}, nil
}

var globalBufferPool bytebufferpool.Pool

type variable struct {
	key      string
	val      string
	res      bool
	deferred bool // this means this variable are undefined right now, and will alloc a temp param and assign it later
}

type taskCall struct {
	name       string
	vars       map[string]*variable
	properties map[string]string
}

func (tc *taskCall) setupTaskName() string { return tc.name + "-setup" }

type UnknownVarError struct {
	Tag string
}

func (e UnknownVarError) Error() string {
	return "UnknownVarError " + e.Tag
}

type SubstituteOpt struct {
	varOnly bool
	noDefer bool
}

func (tc *taskCall) substitute(in string, opt SubstituteOpt) (string, error) {
	buf := globalBufferPool.Get()
	buf.Reset()
	defer globalBufferPool.Put(buf)
	tmpl := fasttemplate.New(in, "{{", "}}")
	_, err := tmpl.ExecuteFunc(buf, func(w io.Writer, tag string) (int, error) {
		val := tc.vars[tag]
		if val == nil || (val.deferred && opt.noDefer) {
			return 0, UnknownVarError{Tag: tag}
		}
		if opt.varOnly && val.res {
			return 0, errors.Errorf("resource %q cannot be used in string context", tag)
		}
		return w.Write([]byte(val.val))
	})
	if err != nil {
		return buf.String(), errors.Wrap(err, fmt.Sprintf("substitute variables in %q", in))
	}
	return buf.String(), nil
}

type planSetupTask struct {
	testbed  *testbedInfo
	template *wf.Template
}

type planTeardownTask struct {
	template *wf.Template
}

type testbedInfo struct {
	inputSpec  json.RawMessage
	outputSpec struct {
		OutputName string              `json:"-"`
		OutputPath string              `json:"output"`
		Items      []testbedOutputItem `json:"items"`
	}
	rmsUrl string
}

type testbedOutputItem struct {
	Name string `json:"name"`
	Path string `json:"path"`
}

func (tb *testbedInfo) argoOutputParams() []wf.Parameter {
	params := make([]wf.Parameter, 0, len(tb.outputSpec.Items)+1)
	// we need the global output here to pass-through testbed output to on-exit handler: https://github.com/argoproj/argo-workflows/blob/20382cab7bbbbd5646a3dbfac6ee18586a6eada5/examples/global-outputs.yaml
	param := wf.Parameter{Name: tb.outputSpec.OutputName, ValueFrom: &wf.ValueFrom{Path: tb.outputSpec.OutputPath}, GlobalName: testbedOutputGlobalName}
	params = append(params, param)
	for _, item := range tb.outputSpec.Items {
		param := wf.Parameter{Name: item.Name, ValueFrom: &wf.ValueFrom{Path: item.Path}}
		params = append(params, param)
	}
	return params
}

func (tb *testbedInfo) inputSpecString() string {
	return string(tb.inputSpec)
}

func (tb *testbedInfo) outputSpecString() string {
	out, _ := json.Marshal(tb.outputSpec)
	return string(out)
}

type compiledPlan struct {
	call     *taskCall
	spec     *TestPlan
	ref      *PlanRef
	template *wf.Template
}

type compiledCase struct {
	call     *taskCall
	spec     *TestCase
	ref      *CaseRef
	template *wf.Template

	timeoutParamName string
}

func buildPlanCall(ctx CompileContext, target *PlanRef, parent *taskCall) (*taskCall, error) {
	spec, err := ctx.PlanSpec(target.PlanID)
	if err != nil {
		return nil, errors.Wrapf(err, "resolve plan #%v", target.PlanID)
	}
	resources := extractResourceNames(spec.TestBed)
	tc := &taskCall{
		name: ctx.GenTemplateName(fmt.Sprintf("plan-%v-", target.PlanID)),
		vars: make(map[string]*variable, len(spec.Variables)+len(resources)),
	}
	for _, decl := range spec.Variables {
		val := decl.Default
		for _, bind := range target.VariableBinds {
			if bind.Name == decl.Name {
				val = bind.Value
				break
			}
		}
		v := &variable{key: decl.Name, val: val, res: false}
		if parent != nil {
			val, err := parent.substitute(val, SubstituteOpt{varOnly: true})
			if err != nil {
				return nil, errors.Wrap(err, fmt.Sprintf("process variable %q of plan %v", decl.Name, target.PlanID))
			}
			v.val = val
		}
		tc.vars["variables."+decl.Name] = v
	}
	return tc, nil
}

// this will be called twice, the first one is just prepare, and second one will be the final version
func buildPlanSetupTask(ctx CompileContext, call *taskCall, spec *TestPlan) (*planSetupTask, error) {
	if spec.TestBed.Name == "" {
		return nil, errors.Errorf("testbed name should be generated before [CompileArgo]")
	}
	testbed, err := extractTestBedInfo(ctx, spec.TestBed, call)
	if err != nil {
		return nil, err
	}
	task := &planSetupTask{}
	task.testbed = testbed
	resReq := corev1.ResourceList{
		corev1.ResourceCPU: kres.MustParse("100m"),
	}
	task.template = &wf.Template{
		Name:    call.name + "-setup",
		Outputs: wf.Outputs{Parameters: testbed.argoOutputParams()},
		Container: &corev1.Container{
			Image:           setupImageParamRefName,
			Resources:       corev1.ResourceRequirements{Requests: resReq},
			ImagePullPolicy: corev1.PullAlways,
			Command:         []string{"/bin/bash", "-c"},
			Args:            nil,
			Env: []corev1.EnvVar{
				{Name: "TESTBED_INPUT_SPEC", Value: testbed.inputSpecString()},
				{Name: "TESTBED_OUTPUT_SPEC", Value: testbed.outputSpecString()},
				{Name: "TOKEN", ValueFrom: &corev1.EnvVarSource{
					SecretKeyRef: &corev1.SecretKeySelector{
						LocalObjectReference: corev1.LocalObjectReference{
							Name: "rms-secret"},
						Key: "token",
					}}},
			},
		},
	}
	task.template.Metadata.Labels = map[string]string{
		"role":           "setUp",
		"planID":         fmtID(spec.ID),
		"rootPlanExecID": fmtID(ctx.RootPlanExecID()),
	}

	var args = make([]string, 0, 1)

	allocCmd := `printenv TESTBED_INPUT_SPEC | tcctl testbed create --wait-timeout=30m --output-spec="${TESTBED_OUTPUT_SPEC}" --credential="" -f -`
	if rms := testbed.rmsUrl; len(rms) > 0 {
		allocCmd = allocCmd + " -r " + rms
	} else {
		return nil, fmt.Errorf("rms addr not configured")
	}

	retryCmd := `total=1
allocated=1
iter=0
while true
do
    %s
    allocated=$?
    iter=$(($iter + 1))
    if [ $allocated = 0 ] || [ $iter -ge $total ]
    then break
    fi
    sleepSeconds=$(( ( $RANDOM %% 10 +  10 * $iter ) * 60 ))
    echo sleep ${sleepSeconds}
    sleep ${sleepSeconds}
done
if [ $allocated != 0 ]
then exit $allocated
fi`
	args = append(args, fmt.Sprintf(retryCmd, allocCmd))

	const localAddr = "localhost:8000"
	const localUrl = "http://" + localAddr + "/api/v1"

	type CurlItem struct {
		name     string
		expr     string
		filePath string
		httpPath string
	}

	toCurlItem := func(expr string, v variable) CurlItem {
		return CurlItem{name: v.key,
			expr:     expr,
			filePath: fmt.Sprintf("/tmp/%s/%s", call.name, v.key),
			httpPath: strings.ReplaceAll(strings.TrimPrefix(expr, "testbed."), ".", "/"),
		}
	}

	curlItems := []CurlItem{}

	for expr, val := range call.vars {
		if val.deferred && !val.res {
			if !(strings.HasPrefix(expr, "testbed.")) {
				return nil, fmt.Errorf("unkown parameter " + expr)
			}
			curlItems = append(curlItems, toCurlItem(expr, *val))
		}
	}

	if len(curlItems) > 0 {
		args = append(args, `set -e`) // fast fail
		args = append(args, fmt.Sprintf("sdkserver -l %s -root-type testbed -handle-file %s &", localAddr, testbed.outputSpec.OutputPath))
		args = append(args, "sleep 2")
	}
	for _, arg := range curlItems {
		args = append(args, fmt.Sprintf("curl --fail %s/%s > %s", localUrl, arg.httpPath, arg.filePath))
		task.template.Outputs.Parameters = append(task.template.Outputs.Parameters, wf.Parameter{Name: arg.name, ValueFrom: &wf.ValueFrom{Path: arg.filePath}})
	}
	if len(curlItems) > 0 {
		args = append(args, "killall sdkserver")
	}
	task.template.Container.Args = []string{strings.Join(args, "\n")}
	return task, nil
}

func buildPlanTeardownTask(ctx CompileContext, testbed *testbedInfo) (*planTeardownTask, error) {
	task := &planTeardownTask{}

	var argumentSuffix string
	if rms := testbed.rmsUrl; len(rms) > 0 {
		argumentSuffix += " -r " + rms
	}
	task.template = &wf.Template{
		Name: teardownTemplateName,
		Container: &corev1.Container{
			Image:           setupImageParamRefName,
			ImagePullPolicy: corev1.PullAlways,
			Command:         []string{"/bin/sh", "-c"},
			Args: []string{fmt.Sprintf(`
if [ {{workflow.status}} = "Succeeded" ]; then
	tcctl testbed complete $(printenv TESTBED_SETUP_OUTPUT | jq '.name' -r) --type Success %s
else
	tcctl testbed complete $(printenv TESTBED_SETUP_OUTPUT | jq '.name' -r) --type Failure %s
fi
`, argumentSuffix, argumentSuffix)},
			Env: []corev1.EnvVar{
				{Name: "TESTBED_SETUP_OUTPUT", Value: fmt.Sprintf("{{workflow.outputs.parameters.%s}}", testbedOutputGlobalName)},
				{Name: "TOKEN", ValueFrom: &corev1.EnvVarSource{
					SecretKeyRef: &corev1.SecretKeySelector{
						LocalObjectReference: corev1.LocalObjectReference{
							Name: "rms-secret"},
						Key: "token",
					}}},
			},
		},
	}
	task.template.Metadata.Labels = map[string]string{
		"role": "tearDown",
	}
	return task, nil
}

func buildDepends(after []string) string {
	var dependencies []string
	for _, s := range after {
		if strings.Contains(s, "||") {
			dependencies = append(dependencies, "("+s+")")
		} else {
			dependencies = append(dependencies, s)
		}
	}
	return strings.Join(dependencies, " && ")
}

func compilePlan(ctx CompileContext, target *PlanRef, parent *taskCall) (*compiledPlan, error) {
	call, err := buildPlanCall(ctx, target, parent)
	if err != nil {
		return nil, err
	}
	spec, err := ctx.PlanSpec(target.PlanID)
	if err != nil {
		return nil, errors.Wrapf(err, "resolve plan #%v", target.PlanID)
	}

	pipeline, err := compilePipeline(ctx, spec, call)
	if err != nil {
		return nil, err
	}
	ctx.AddTemplate(pipeline)

	setup, err := buildPlanSetupTask(ctx, call, spec)
	if err != nil {
		return nil, err
	}
	ctx.AddTemplate(setup.template)
	teardown, err := buildPlanTeardownTask(ctx, setup.testbed)
	if err != nil {
		return nil, err
	}
	ctx.AddTemplate(teardown.template)

	var pipelineBinds []wf.Parameter
	for _, v := range call.vars {
		if v.deferred {
			pipelineBinds = append(pipelineBinds,
				wf.Parameter{
					Name:  v.key,
					Value: wf.AnyStringPtr(fmt.Sprintf("{{tasks.%s.outputs.parameters.%s}}", setup.template.Name, v.key)),
				})
		}
	}

	dag := &wf.DAGTemplate{
		Tasks: []wf.DAGTask{
			{Name: setup.template.Name, Template: setup.template.Name},
			{
				Name:         pipeline.Name,
				Template:     pipeline.Name,
				Arguments:    wf.Arguments{Parameters: pipelineBinds},
				Dependencies: []string{setup.template.Name},
			},
		},
		Target: pipeline.Name,
	}
	template := &wf.Template{Name: call.name, DAG: dag}
	ctx.AddTemplate(template)

	return &compiledPlan{
		call:     call,
		spec:     spec,
		ref:      target,
		template: template,
	}, nil
}

func buildPipelineCall(ctx CompileContext, spec *TestPlan, parent *taskCall) (*taskCall, error) {
	tc := &taskCall{
		name:       "pipeline",
		vars:       make(map[string]*variable, len(parent.vars)),
		properties: parent.properties,
	}
	for k, v := range parent.vars {
		tc.vars[k] = v
	}
	resources := extractResourceNames(spec.TestBed)
	for _, name := range resources {
		tc.vars["testbed."+name] = &variable{
			key:      name,
			val:      fmt.Sprintf("{{inputs.parameters.%s}}", name),
			res:      true,
			deferred: true,
		}
	}
	return tc, nil
}

func compilePipeline(ctx CompileContext, spec *TestPlan, planCall *taskCall) (*wf.Template, error) {
	tc, err := buildPipelineCall(ctx, spec, planCall)
	if err != nil {
		return nil, err
	}
	template := &wf.Template{}
	if spec.Pipeline != nil {
		dag, err := compileDAG(ctx, spec, tc)
		if err != nil {
			return nil, err
		}
		template.DAG = dag
	} else if spec.Steps != nil {
		steps, err := compileSteps(ctx, spec, tc)
		if err != nil {
			return nil, err
		}
		template.Steps = steps
	}
	template.Name = tc.name
	for k, v := range tc.vars {
		if v.deferred {
			planCall.vars[k] = v
			param := wf.Parameter{Name: v.key}
			template.Inputs.Parameters = append(template.Inputs.Parameters, param)
		}
	}
	return template, nil
}

func compileSteps(ctx CompileContext, plan *TestPlan, call *taskCall) ([]wf.ParallelSteps, error) {
	call.name = "steps"
	var steps []wf.ParallelSteps
	var continueOn *wf.ContinueOn
	if !plan.FailFast {
		continueOn = &wf.ContinueOn{Failed: true}
	}
	for _, step := range plan.Steps {
		task, err := compileCaseInvoke(ctx, step, call, plan)
		if err != nil {
			return nil, err
		}
		step := wf.WorkflowStep{Name: task.Name, Template: task.Template, Arguments: task.Arguments, When: task.When, ContinueOn: continueOn}
		sg := wf.ParallelSteps{Steps: []wf.WorkflowStep{step}}
		steps = append(steps, sg)
	}
	if !plan.FailFast && steps != nil {
		template := newStepsFinalizerTemplate()
		ctx.AddTemplate(&template)
		steps = append(steps, genStepsFinalizerStep(steps, template))
	}
	return steps, nil
}

func newStepsFinalizerTemplate() wf.Template {
	return wf.Template{
		Name:   "steps-phase-finalizer",
		Inputs: wf.Inputs{Parameters: []wf.Parameter{{Name: "phases"}}},
		Script: &wf.ScriptTemplate{
			Container: corev1.Container{Image: "alpine", Command: []string{"sh"}},
			Source: `for s in {{inputs.parameters.phases}}; do
if [ $s = Failed ] || [ $s = Error ]; then
	exit 1
fi
done`,
		}}
}

func genStepsFinalizerStep(steps []wf.ParallelSteps, finalizer wf.Template) wf.ParallelSteps {
	var items []string
	for _, sg := range steps {
		for _, step := range sg.Steps {
			items = append(items, fmt.Sprintf("{{steps.%s.status}}", step.Name))
		}
	}
	expr := strings.Join(items, " ")
	step := wf.WorkflowStep{
		Name:     "steps-done",
		Template: finalizer.Name,
		Arguments: wf.Arguments{
			Parameters: []wf.Parameter{{
				Name:  finalizer.Inputs.Parameters[0].Name,
				Value: wf.AnyStringPtr(expr)},
			},
		},
	}
	return wf.ParallelSteps{Steps: []wf.WorkflowStep{step}}
}

func compileDAG(ctx CompileContext, spec *TestPlan, call *taskCall) (*wf.DAGTemplate, error) {
	call.name = "pipeline"
	dag := &wf.DAGTemplate{
		Tasks:    make([]wf.DAGTask, 0, len(spec.Pipeline)+2),
		FailFast: &spec.FailFast,
	}
	var cases []string
	for _, step := range spec.Pipeline {
		task, err := compileCaseInvoke(ctx, step, call, spec)
		if err != nil {
			return nil, err
		}
		cases = append(cases, task.Name)
		dag.Tasks = append(dag.Tasks, *task)
	}
	dag.Target = strings.Join(cases, " ")
	return dag, nil
}

func compileCaseInvoke(ctx CompileContext, step PipelineStep, call *taskCall, plan *TestPlan) (*wf.DAGTask, error) {
	when, err := buildCaseWhen(ctx, &step.CaseRef, plan)
	if err != nil {
		when = "" // do nothing if version error
	}
	task := &wf.DAGTask{
		Name:    step.Name,
		When:    when,
		Depends: buildDepends(step.After),
	}
	child, err := compileCase(ctx, &step, call)
	if err != nil {
		return nil, err
	}
	task.Template = child.template.Name
	for _, bind := range step.CaseRef.Arguments {
		for _, val := range child.call.vars {
			if val.key == bind.Name && val.deferred {
				var rendered string
				var err error
				const MaxSubstitute = 999
				var trial int = 0
				for ; trial < MaxSubstitute; trial++ {
					rendered, err = call.substitute(bind.Value, SubstituteOpt{varOnly: false})
					e := UnknownVarError{}
					if err != nil && errors.As(err, &e) {
						if strings.HasPrefix(e.Tag, "testbed.items.") {
							tmpName := fmt.Sprintf("__param_%d", ctx.NextUniqueInt())
							call.vars[e.Tag] = &variable{
								key:      tmpName,
								val:      fmt.Sprintf("{{inputs.parameters.%s}}", tmpName),
								res:      false,
								deferred: true,
							}
						}
						continue
					}
					break
				}
				if err != nil {
					return nil, err
				}
				if trial == MaxSubstitute {
					return nil, fmt.Errorf("maxSubstitute exceeded")
				}
				param := wf.Parameter{Name: bind.Name, Value: wf.AnyStringPtr(rendered)}
				task.Arguments.Parameters = append(task.Arguments.Parameters, param)
			}
		}
	}
	if step.Timeout > 0 {
		param := wf.Parameter{
			Name:  child.timeoutParamName,
			Value: wf.AnyStringPtr((time.Duration(step.Timeout) * time.Second).String()),
		}
		task.Arguments.Parameters = append(task.Arguments.Parameters, param)
	}
	return task, nil
}

func buildCaseWhen(ctx CompileContext, target *CaseRef, plan *TestPlan) (when string, err error) {
	supported, err := isCaseVersionSupported(ctx, target, plan)
	if err != nil {
		return "", err
	}
	if !supported {
		return "false", nil
	}
	return "", nil
}

func extractPlanTidbVersionStr(plan *TestPlan) (ver string) {
	ver = plan.GitHubPlanInfo.Branch
	if ver != "" {
		return ver
	}
	_, right, found := strings.Cut(plan.Name, "@")
	if !found {
		return ""
	}
	return right
}

func isCaseVersionSupported(ctx CompileContext, target *CaseRef, plan *TestPlan) (supported bool, err error) {
	spec, err := ctx.CaseSpec(target.CaseID)
	if err != nil {
		return
	}
	tidbVersionStr := extractPlanTidbVersionStr(plan)
	if tidbVersionStr == "" {
		return true, nil
	}
	// release means latest, hack set it 9999 as largest patch version
	if strings.HasPrefix(tidbVersionStr, "release-") {
		tidbVersionStr = fmt.Sprintf("%s.9999", strings.TrimPrefix(tidbVersionStr, "release-"))
	}
	tidbVersion, err := versionlib.NewVersion(tidbVersionStr)
	if err != nil {
		return
	}
	// ignore pre version because of bad schema
	if tidbVersion.Prerelease() != "" {
		tidbVersionStr = strings.TrimSuffix(tidbVersionStr, "-"+tidbVersion.Prerelease())
		tidbVersion, err = versionlib.NewVersion(tidbVersionStr)
		if err != nil {
			return
		}
	}
	if len(spec.SupportedVersions) == 0 {
		return true, nil
	}
	for _, version := range spec.SupportedVersions {
		constraint, err := versionlib.NewConstraint(version)
		if err != nil {
			return false, err
		}
		if constraint.Check(tidbVersion) {
			return true, nil
		}
	}
	return false, nil
}

func buildCaseCall(ctx CompileContext, target *CaseRef, parent *taskCall) (*taskCall, error) {
	spec, err := ctx.CaseSpec(target.CaseID)
	if err != nil {
		return nil, errors.Wrapf(err, "resolve case #%v", target.CaseID)
	}
	if spec == nil {
		return nil, errors.Errorf("no such case #%v", target.CaseID)
	}
	tc := &taskCall{
		name:       ctx.GenTemplateName(fmt.Sprintf("case-%v-", target.CaseID)),
		vars:       make(map[string]*variable, len(spec.Parameters)),
		properties: parent.properties,
	}

	for _, decl := range spec.Parameters {
		valTmpl := decl.Default
		for _, bind := range target.Arguments {
			if bind.Name == decl.Name {
				valTmpl = bind.Value
				break
			}
		}
		v := &variable{key: decl.Name}
		if decl.Type == CaseParamTypeResource {
			v.res = true
			v.deferred = true
		} else {
			v.val = valTmpl
			if parent != nil {
				val, err := parent.substitute(v.val, SubstituteOpt{varOnly: true, noDefer: true})
				if err != nil {
					unknownVar := UnknownVarError{}
					if errors.As(err, &unknownVar) {
						v.deferred = true
					} else {
						return nil, errors.Wrap(err, fmt.Sprintf("process parameter %q of case %v", decl.Name, target.CaseID))
					}
				}
				v.val = val
			}
		}
		if v.deferred {
			v.val = fmt.Sprintf("{{inputs.parameters.%s}}", decl.Name)
		}
		tc.vars["parameters."+decl.Name] = v
	}
	return tc, nil
}

func compileCase(ctx CompileContext, step *PipelineStep, parent *taskCall) (*compiledCase, error) {
	call, err := buildCaseCall(ctx, &step.CaseRef, parent)
	if err != nil {
		return nil, err
	}
	spec, err := ctx.CaseSpec(step.CaseID)
	if err != nil {
		return nil, errors.Wrapf(err, "resolve case #%v", step.CaseID)
	}
	info := spec.Framework.Generic
	if info == nil {
		return nil, errors.Errorf("case #%v: framework.generic is nil", step.CaseID)
	}

	template := &wf.Template{}
	container := &corev1.Container{}
	// image
	caseImage := info.Image
	if step.CaseImage != "" {
		// rewrite image by plan step
		caseImage = step.CaseImage
	}
	image, err := call.substitute(caseImage, SubstituteOpt{varOnly: true})
	if err != nil {
		return nil, err
	}
	container.Image = image
	// imagePullPolicy
	imagePullPolicy, err := call.substitute(string(info.ImagePullPolicy), SubstituteOpt{varOnly: true})
	if err != nil {
		return nil, err
	}
	container.ImagePullPolicy = corev1.PullPolicy(imagePullPolicy)
	// workingDir
	workingDir, err := call.substitute(info.WorkingDir, SubstituteOpt{varOnly: true})
	if err != nil {
		return nil, err
	}
	container.WorkingDir = workingDir
	// command
	command := make([]string, len(info.Command))
	for i, item := range info.Command {
		processed, err := call.substitute(item, SubstituteOpt{varOnly: false})
		if err != nil {
			return nil, err
		}
		command[i] = processed
	}
	container.Command = command
	// env
	envNames := make([]string, 0, 8)
	env := make(map[string]string)
	for k, v := range info.Env {
		processed, err := call.substitute(v, SubstituteOpt{varOnly: false})
		if err != nil {
			return nil, err
		}
		env[k] = processed
		envNames = append(envNames, k)
	}
	for _, v := range call.vars {
		prefix := common.TestEnvStrParamPrefix
		if v.res {
			prefix = common.TestEnvResParamPrefix
		}
		if v.deferred {
			template.Inputs.Parameters = append(template.Inputs.Parameters, wf.Parameter{Name: v.key})
		}
		k := prefix + v.key
		env[k] = v.val
		envNames = append(envNames, k)
	}
	caseExecID, err := ctx.NextCaseExecID(step)
	if err != nil {
		return nil, err
	}
	env[common.TestEnvCaseExecID] = fmtID(caseExecID)
	envNames = append(envNames, common.TestEnvCaseExecID)
	archivePath := path.Join("/tmp", call.name)
	outputPath := path.Join("/tmp", call.name+".out")
	env[common.TestEnvArchivePath] = archivePath
	env[common.TestEnvOutputPath] = outputPath
	envNames = append(envNames, common.TestEnvArchivePath, common.TestEnvOutputPath)

	sort.Strings(envNames)
	for _, name := range envNames {
		container.Env = append(container.Env, corev1.EnvVar{Name: name, Value: env[name]})
	}

	template.Name = call.name
	template.Container = container

	// timeout
	timeoutParamName := call.name + "-timeout"
	timeoutParam := wf.Parameter{Name: timeoutParamName}
	if timeout := info.DefaultTimeout; timeout > 0 {
		timeoutParam.Default = wf.AnyStringPtr((time.Duration(timeout) * time.Second).String())
	} else {
		timeoutParam.Default = wf.AnyStringPtr("")
	}
	template.Inputs.Parameters = append(template.Inputs.Parameters, timeoutParam)

	archiveArtifact := wf.Artifact{Name: "archive", Path: archivePath, Optional: true}
	outputArtifact := wf.Artifact{Name: "output", Path: outputPath, Optional: true, Archive: &wf.ArchiveStrategy{None: &wf.NoneStrategy{}}}
	template.Outputs.Artifacts = append(template.Outputs.Artifacts, archiveArtifact, outputArtifact)
	template.Timeout = "{{inputs.parameters." + timeoutParamName + "}}"
	template.Metadata.Labels = map[string]string{
		"role":           "case",
		"caseID":         fmtID(step.CaseID),
		"caseExecID":     fmtID(caseExecID),
		"rootPlanExecID": fmtID(ctx.RootPlanExecID()),
	}

	ctx.AddTemplate(template)

	return &compiledCase{
		call:             call,
		spec:             spec,
		ref:              &step.CaseRef,
		template:         template,
		timeoutParamName: timeoutParamName,
	}, nil
}

func extractResourceNames(testbed TestBed) []string {
	names := make([]string, len(testbed.Items))
	for i, item := range testbed.Items {
		names[i] = item.Name
	}
	return names
}

func extractTestBedInfo(ctx CompileContext, testbed TestBed, call *taskCall) (*testbedInfo, error) {
	raw, err := json.Marshal(testbed)
	if err != nil {
		return nil, errors.Wrap(err, "encode testbed")
	}
	var obj map[string]interface{}
	err = json.Unmarshal(raw, &obj)
	if err != nil {
		return nil, errors.Wrap(err, "decode testbed")
	}
	processed, err := recursiveSubstitute(obj, func(s string) (string, error) { return call.substitute(s, SubstituteOpt{varOnly: true}) })
	if err != nil {
		return nil, err
	}
	attachMetaInfo(ctx, processed.(map[string]interface{}))
	input, err := json.Marshal(processed)
	if err != nil {
		return nil, errors.Wrap(err, "encode processed testbed")
	}
	info := &testbedInfo{inputSpec: input}
	info.outputSpec.OutputName = call.name + "-setup-output"
	info.outputSpec.OutputPath = fmt.Sprintf("/tmp/%s/output", call.name)
	info.outputSpec.Items = make([]testbedOutputItem, len(testbed.Items))
	for i, item := range testbed.Items {
		info.outputSpec.Items[i] = testbedOutputItem{
			Name: item.Name,
			Path: fmt.Sprintf("/tmp/%s/items/%s", call.name, item.Name),
		}
	}
	info.rmsUrl = testbed.RmsUrl
	return info, nil
}

func attachMetaInfo(ctx CompileContext, testbed map[string]interface{}) map[string]interface{} {
	m := request.TestBedMeta{
		AuthorizeTo: []request.User{{Kind: "ServiceAccount", Name: "tipocket-runner", Namespace: "testground"}},
		Accounting:  request.User{Name: fmtID(ctx.RootPlanExecID()), Kind: "ID"},
	}
	testbed["meta"] = m
	return testbed
}

func recursiveSubstitute(x interface{}, substitute func(string) (string, error)) (interface{}, error) {
	if substitute == nil {
		return x, nil
	}
	switch v := x.(type) {
	case string:
		out, err := substitute(v)
		if err != nil {
			return nil, err
		}
		return out, nil
	case map[string]interface{}:
		out := make(map[string]interface{}, len(v))
		for k, x := range v {
			y, err := recursiveSubstitute(x, substitute)
			if err != nil {
				return nil, err
			}
			out[k] = y
		}
		return out, nil
	case []interface{}:
		out := make([]interface{}, len(v))
		for i, x := range v {
			y, err := recursiveSubstitute(x, substitute)
			if err != nil {
				return nil, err
			}
			out[i] = y
		}
		return out, nil
	default:
		return x, nil
	}
}

func fmtID(id ID) string {
	return strconv.FormatUint(uint64(id), 10)
}