package schedule

import (
	"context"
	"fmt"
	"net/url"

	wfApiClient "github.com/argoproj/argo-workflows/v3/pkg/apiclient"
	workflowpkg "github.com/argoproj/argo-workflows/v3/pkg/apiclient/workflow"
	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	"github.com/pingcap/test-infra/services/tps/pkg/common"
)

type ArgoClientMgr interface {
	ClientFor(job *common.ArgoJob) (ArgoClient, error)
	BindClient(job *common.ArgoJob, client ArgoClient) *common.ArgoJob
}

type ArgoClient interface {
	Create(ctx context.Context, workflow *wfapi.Workflow) (*wfapi.Workflow, error)
	Get(ctx context.Context, name string) (*wfapi.Workflow, error)
	Stop(ctx context.Context, name string) (*wfapi.Workflow, error)
	URL() string
}

type argoClient struct {
	config    ArgoConfig
	wfClient  workflowpkg.WorkflowServiceClient
	clientCtx context.Context
}

func (c argoClient) Create(ctx context.Context, workflow *wfapi.Workflow) (*wfapi.Workflow, error) {
	return c.wfClient.CreateWorkflow(
		c.clientCtx,
		&workflowpkg.WorkflowCreateRequest{Namespace: c.config.Namespace, Workflow: decorateWorkflow(workflow, c.config)},
	)
}

func decorateWorkflow(workflow *wfapi.Workflow, config ArgoConfig) *wfapi.Workflow {
	workflow.Spec.ArtifactRepositoryRef = config.ArtifactRepoRef
	workflow.Spec.ServiceAccountName = config.ServiceAccount
	workflow.Namespace = config.Namespace
	if !isSetupImageParamExist(workflow.Spec.Arguments.Parameters) {
		arg := wfapi.Parameter{Name: setupImageParamName, Value: wfapi.AnyStringPtr(config.SetupImage)}
		workflow.Spec.Arguments.Parameters = append(workflow.Spec.Arguments.Parameters, arg)
	}
	return workflow
}

func isSetupImageParamExist(args []wfapi.Parameter) bool {
	for _, arg := range args {
		if arg.Name == setupImageParamName {
			return true
		}
	}
	return false
}

func (c argoClient) Get(ctx context.Context, name string) (*wfapi.Workflow, error) {
	return c.wfClient.GetWorkflow(c.clientCtx, &workflowpkg.WorkflowGetRequest{Namespace: c.config.Namespace, Name: name})
}

func (c argoClient) Stop(ctx context.Context, name string) (*wfapi.Workflow, error) {
	return c.wfClient.StopWorkflow(c.clientCtx, &workflowpkg.WorkflowStopRequest{Namespace: c.config.Namespace, Name: name})
}

func (c argoClient) URL() string {
	return c.config.URL
}

type ArgoConfig struct {
	ServiceAccount  string                       `json:"serviceAccount,omitempty"`
	ArtifactRepoRef *wfapi.ArtifactRepositoryRef `json:"artifactRepoRef"`
	Namespace       string                       `json:"namespace,omitempty"`
	URL             string                       `json:"url,omitempty"`
	SetupImage      string                       `json:"setupImage"`
}

func ArgoClientFromConfig(cfg ArgoConfig) (*argoClient, error) {
	parsed, err := url.Parse(cfg.URL)
	if err != nil {
		return nil, err
	}
	secure := false
	if parsed.Scheme == "https" {
		secure = true
	}
	clientCtx, client, err := wfApiClient.NewClientFromOpts(
		wfApiClient.Opts{
			ArgoServerOpts: wfApiClient.ArgoServerOpts{
				URL:    parsed.Host,
				Path:   parsed.Path,
				Secure: secure,
				HTTP1:  true,
			},
			AuthSupplier: func() string {
				return ""
			},
		})
	if err != nil {
		return nil, err
	}
	return &argoClient{config: cfg, clientCtx: clientCtx, wfClient: client.NewWorkflowServiceClient()}, nil
}

type ArgoByRms struct {
	pools    map[string]*argoClient
	selector map[string]string
}

const ArgoClientKey = "argo.svc.url"

func (a ArgoByRms) ClientFor(job *common.ArgoJob) (ArgoClient, error) {
	var clientKey string
	if key := job.PlanExec.Meta[ArgoClientKey]; key != "" {
		clientKey = key
	} else {
		clientKey = a.selector[job.Plan.TestBed.RmsUrl]
	}
	client, ok := a.pools[clientKey]
	if !ok {
		return nil, fmt.Errorf("argo client not found for: %s", clientKey)
	}
	return client, nil
}

func (a ArgoByRms) BindClient(job *common.ArgoJob, client ArgoClient) *common.ArgoJob {
	if job.PlanExec.Meta == nil {
		job.PlanExec.Meta = map[string]string{}
	}
	job.PlanExec.Meta[ArgoClientKey] = client.URL()
	return job
}

func NewArgoByRms(configs []ArgoConfig, selector map[string]string) (*ArgoByRms, error) {
	pools := make(map[string]*argoClient, len(configs))
	for _, config := range configs {
		client, err := ArgoClientFromConfig(config)
		if err != nil {
			return nil, err
		}
		pools[client.URL()] = client
	}
	return &ArgoByRms{selector: selector, pools: pools}, nil
}