package workflow

import (
	"context"
	"fmt"
	"k8s.io/apimachinery/pkg/api/errors"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"strconv"

	"k8s.io/apimachinery/pkg/types"

	"github.com/Azure/Orkestra/pkg/meta"
	v1alpha13 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/Azure/Orkestra/api/v1alpha1"
	"github.com/go-logr/logr"
)

var _ = ForwardWorkflowClient{}
var _ = ReverseWorkflowClient{}
var _ = RollbackWorkflowClient{}

type Client interface {
	// Generate the object required by the workflow engine
	Generate(ctx context.Context) error

	// Submit the object required by the workflow engine generated by the Generate method
	Submit(ctx context.Context) error

	// GetType returns the workflow client type
	GetType() v1alpha1.WorkflowType

	// GetLogger returns the logger associated with the workflow client
	GetLogger() logr.Logger

	// GetName returns the name associated with the workflow
	GetName() string

	// GetNamespace returns the namespace that the workflow should run in
	GetNamespace() string

	// GetOptions returns the client options used with the workflow client
	GetOptions() ClientOptions

	// GetClient returns the k8s client associated with the workflow
	GetClient() client.Client

	// GetWorkflow returns the workflow associated with the client
	GetWorkflow() *v1alpha13.Workflow

	// GetAppGroup returns the app group from the workflow client
	GetAppGroup() *v1alpha1.ApplicationGroup
}

type ClientOptions struct {
	Parallelism *int64
	StagingRepo string
	Namespace   string
}

type Builder struct {
	client  client.Client
	options ClientOptions
	logger  logr.Logger

	workflow *v1alpha13.Workflow
	appGroup *v1alpha1.ApplicationGroup
}

type ForwardWorkflowClient struct {
	client.Client
	logr.Logger
	ClientOptions

	workflow *v1alpha13.Workflow
	appGroup *v1alpha1.ApplicationGroup
}

type RollbackWorkflowClient struct {
	client.Client
	logr.Logger
	ClientOptions

	workflow *v1alpha13.Workflow
	appGroup *v1alpha1.ApplicationGroup
}

type ReverseWorkflowClient struct {
	client.Client
	logr.Logger
	ClientOptions

	workflow *v1alpha13.Workflow
	appGroup *v1alpha1.ApplicationGroup
}

func NewBuilder(client client.Client, logger logr.Logger) *Builder {
	return &Builder{
		client:  client,
		options: ClientOptions{},
		logger:  logger,
	}
}

func NewClientFromClient(client Client, clientType v1alpha1.WorkflowType) Client {
	builder := &Builder{
		client:  client.GetClient(),
		options: client.GetOptions(),
		logger:  client.GetLogger(),
	}
	return builder.Build(clientType, client.GetAppGroup())
}

func (builder *Builder) WithParallelism(numNodes int64) *Builder {
	builder.options.Parallelism = &numNodes
	return builder
}

func (builder *Builder) WithStagingRepo(stagingURL string) *Builder {
	builder.options.StagingRepo = stagingURL
	return builder
}

func (builder *Builder) InNamespace(namespace string) *Builder {
	builder.options.Namespace = namespace
	return builder
}

func (builder *Builder) Build(clientType v1alpha1.WorkflowType, appGroup *v1alpha1.ApplicationGroup) Client {
	switch clientType {
	case v1alpha1.Forward:
		forwardClient := &ForwardWorkflowClient{
			Client:        builder.client,
			Logger:        builder.logger,
			ClientOptions: builder.options,
			appGroup:      appGroup,
		}
		return forwardClient
	case v1alpha1.Reverse:
		reverseClient := &ReverseWorkflowClient{
			Client:        builder.client,
			Logger:        builder.logger,
			ClientOptions: builder.options,
			appGroup:      appGroup,
		}
		return reverseClient
	default:
		rollbackClient := &RollbackWorkflowClient{
			Client:        builder.client,
			Logger:        builder.logger,
			ClientOptions: builder.options,
			appGroup:      appGroup,
		}
		return rollbackClient
	}
}

// Run calls the generate and Submit commands of the workflow client
func Run(ctx context.Context, wfClient Client) error {
	if err := wfClient.Generate(ctx); err != nil {
		return fmt.Errorf("failed to generate workflow: %w", err)
	}
	if err := wfClient.Submit(ctx); err != nil {
		return fmt.Errorf("failed to submit workflow: %w", err)
	}
	return nil
}

// Submit calls the base submit function for the workflow client
func Submit(ctx context.Context, wfClient Client) error {
	controllerutil.AddFinalizer(wfClient.GetWorkflow(), v1alpha1.AppGroupFinalizer)
	wfClient.GetWorkflow().GetLabels()[v1alpha1.OwnershipLabel] = wfClient.GetAppGroup().Name
	wfClient.GetWorkflow().GetLabels()[v1alpha1.WorkflowAppGroupGenerationLabel] = strconv.FormatInt(wfClient.GetAppGroup().Generation, 10)

	if err := wfClient.GetClient().Create(ctx, wfClient.GetWorkflow()); !errors.IsAlreadyExists(err) && err != nil {
		return fmt.Errorf("failed to CREATE argo workflow object: %w", err)
	} else if errors.IsAlreadyExists(err) {
		// If the workflow needs an update, delete the previous workflow and apply the new one
		// Argo Workflow does not rerun the workflow on UPDATE, so instead we cleanup and re-apply
		if err := DeleteWorkflow(ctx, wfClient); err != nil {
			return fmt.Errorf("failed to DELETE argo workflow object: %w", err)
		}
		if err := controllerutil.SetControllerReference(wfClient.GetAppGroup(), wfClient.GetWorkflow(), wfClient.GetClient().Scheme()); err != nil {
			return fmt.Errorf("unable to set ApplicationGroup as owner of Argo Workflow: %w", err)
		}
		if err := wfClient.GetClient().Create(ctx, wfClient.GetWorkflow()); err != nil {
			return fmt.Errorf("failed to CREATE argo workflow object: %w", err)
		}
	}
	return nil
}

// Suspend sets the suspend flag on the workflow associated with the workflow client
// if the workflow still exists on the cluster
func Suspend(ctx context.Context, wfClient Client) error {
	// suspend a workflow if it is not already finished or suspended
	workflow, err := GetWorkflow(ctx, wfClient)
	if client.IgnoreNotFound(err) != nil {
		return fmt.Errorf("failed to suspend the workflow: %w", err)
	} else if err != nil || !workflow.Status.FinishedAt.IsZero() {
		wfClient.GetLogger().Info("workflow not found, no need to suspend")
		return nil
	}
	if workflow.Spec.Suspend == nil || !*workflow.Spec.Suspend {
		wfClient.GetLogger().Info("suspending the workflow")
		patch := client.MergeFrom(workflow.DeepCopy())
		suspend := true
		workflow.Spec.Suspend = &suspend
		if err := wfClient.GetClient().Patch(ctx, workflow, patch); err != nil {
			return fmt.Errorf("failed to patch the workflow: %w", err)
		}
		SetSuspended(wfClient)
	}
	return nil
}

func GetWorkflow(ctx context.Context, wc Client) (*v1alpha13.Workflow, error) {
	workflow := &v1alpha13.Workflow{}
	err := wc.GetClient().Get(ctx, types.NamespacedName{Namespace: wc.GetNamespace(), Name: wc.GetName()}, workflow)
	return workflow, err
}

// DeleteWorkflow removes the workflow from the api server associated with
// the workflow client
func DeleteWorkflow(ctx context.Context, wfClient Client) error {
	workflow := &v1alpha13.Workflow{}
	if err := wfClient.GetClient().Get(ctx, types.NamespacedName{Name: wfClient.GetName(), Namespace: wfClient.GetNamespace()}, workflow); err != nil {
		return err
	}
	deletePropagation := metav1.DeletePropagationForeground
	return wfClient.GetClient().Delete(ctx, workflow, &client.DeleteOptions{PropagationPolicy: &deletePropagation})
}

// SetProgressing sets one of the workflow conditions in the progressing state
func SetProgressing(parent *v1alpha1.ApplicationGroup, wfType v1alpha1.WorkflowType) {
	if condition, ok := v1alpha1.WorkflowConditionMap[wfType]; ok {
		meta.SetResourceCondition(parent, condition, metav1.ConditionUnknown, meta.ProgressingReason, "workflow is progressing...")
	}
}

// SetSucceeded sets one of the workflow conditions in the succeeded state
func SetSucceeded(parent *v1alpha1.ApplicationGroup, wfType v1alpha1.WorkflowType) {
	if condition, ok := v1alpha1.WorkflowConditionMap[wfType]; ok {
		meta.SetResourceCondition(parent, condition, metav1.ConditionTrue, meta.SucceededReason, "workflow succeeded")
	}
}

// SetFailed sets one of the workflow conditions in the failed state
func SetFailed(parent *v1alpha1.ApplicationGroup, wfType v1alpha1.WorkflowType, message string) {
	if condition, ok := v1alpha1.WorkflowConditionMap[wfType]; ok {
		meta.SetResourceCondition(parent, condition, metav1.ConditionFalse, meta.FailedReason, message)
	}
}

// SetSuspended sets one of the workflow conditions in the suspended state
func SetSuspended(wfClient Client) {
	if condition, ok := v1alpha1.WorkflowConditionMap[wfClient.GetType()]; ok {
		meta.SetResourceCondition(wfClient.GetAppGroup(), condition, metav1.ConditionFalse, meta.SuspendedReason, "workflow is suspended")
	}
}

// IsFailed checks if the workflow created by the workflow client is in a failed state
func IsFailed(ctx context.Context, wfClient Client) (bool, error) {
	wf, err := GetWorkflow(ctx, wfClient)
	if client.IgnoreNotFound(err) != nil {
		return false, fmt.Errorf("failed to get workflow: %w", err)
	}
	if ToConditionReason(wf.Status.Phase) == meta.FailedReason {
		return true, nil
	}
	return false, nil
}

// IsSucceeded checks if the workflow created by the workflow client is in a succeeded state
func IsSucceeded(ctx context.Context, wfClient Client) (bool, error) {
	wf, err := GetWorkflow(ctx, wfClient)
	if client.IgnoreNotFound(err) != nil {
		return false, fmt.Errorf("failed to get workflow: %w", err)
	}
	if ToConditionReason(wf.Status.Phase) == meta.SucceededReason {
		return true, nil
	}
	return false, nil
}

func ToConditionReason(nodePhase v1alpha13.WorkflowPhase) string {
	switch nodePhase {
	case v1alpha13.WorkflowFailed:
		return meta.FailedReason
	case v1alpha13.WorkflowSucceeded:
		return meta.SucceededReason
	default:
		return meta.ProgressingReason
	}
}
