// Package dm add by blacknc <blacknc@163.com> at 2024/09/02 18:27:06
package dm

import (
	"context"
	"time"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/internal/handler/agent"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/util/wait"
)

const (
	OperationTimeout = 5 * time.Second
	CheckInterval    = 500 * time.Millisecond
)

var Tracer = otel.Tracer("central/model/dm")

type ResourceAPI interface {
	CreateDMJobIgnoreExists(ctx context.Context, dmJob *v1alpha1.DMJob) error
	UpdateDMJob(ctx context.Context, dmJob *v1alpha1.DMJob) error
	GetDMJob(ctx context.Context, projectID, dmJobID uint64) (*v1alpha1.DMJob, error)
	DeleteDMJob(ctx context.Context, projectID, dmJobID uint64) (err error)
	GetObjectNamespace(projectID uint64) string
	IsNotFound(error) bool
}

func dmGetObjectNamespace(projectID uint64) string {
	return infraapi.GetObjectNamespace(projectID)
}

func NewRegionalResourceAPI(ctx context.Context, provider, region string, tenantID uint64) (ResourceAPI, error) {
	regionalAPIEndpoint, err := agent.BuildRegionalAPIEndpoint(ctx, provider, region, tenantID)
	if err != nil {
		log.WithContextL(ctx).Error("failed to build regional api endpoint", zap.Error(err))
		return nil, err
	}
	return &regionalResourceAPI{
		regionalAPI: regionalapi.NewTCIDMHTTPClient(regionalAPIEndpoint),
	}, nil
}

type regionalResourceAPI struct {
	regionalAPI regionalapi.TCIDMHTTPClient
}

func (a *regionalResourceAPI) DeleteDMJob(ctx context.Context, projectID, dmJobID uint64) (err error) {
	ctx, span := Tracer.Start(ctx, "RegionalAPI-DeleteDMJob")
	defer func() { tracing.EndSpan(span, err) }()

	ns := a.GetObjectNamespace(projectID)
	name := getDMJobName(dmJobID)
	span.SetAttributes(attribute.String("dm-job-name", name), attribute.String("namespace", ns))

	err = a.regionalAPI.DelDMJobIgnoreNotExists(ctx, ns, name)
	if err != nil {
		return errors.Annotatef(err, "delete dmJob CR %q in namespace %q failed", name, ns)
	}

	// Added at 20250321 by fangdong.jia@pingcap.com
	// Due to the addition of delayed resource release logic in the dm controller,
	// the CR is marked for deletion and exits directly without waiting for complete deletion
	//
	// return wait.PollUntilContextTimeout(ctx, CheckInterval, OperationTimeout, true, func(ctx context.Context) (bool, error) {
	// 	imp, err := a.GetDMJob(ctx, projectID, dmJobID)
	// 	if err != nil && !regionalapi.TCIDMIsNotFound(err) && imp == nil {
	// 		return false, err
	// 	}
	// 	return imp == nil, nil
	// })
	return nil
}

func (a *regionalResourceAPI) CreateDMJobIgnoreExists(ctx context.Context, dmJob *v1alpha1.DMJob) (err error) {

	ctx, span := Tracer.Start(ctx, "RegionalAPI-CreateDMJobIgnoreExists")
	defer func() { tracing.EndSpan(span, err) }()

	span.SetAttributes(attribute.String("dm-job-name", dmJob.Name), attribute.String("namespace", dmJob.Namespace))

	err = wait.PollUntilContextTimeout(ctx, CheckInterval, OperationTimeout, true, func(ctx context.Context) (bool, error) {
		err := a.regionalAPI.CreateDMJobIgnoreExists(ctx, dmJob)
		if err != nil {
			return false, nil
		}
		return true, nil
	})
	if err != nil {
		return errors.Annotatef(err, "create dmJob CR %q for cluster %q in namespace %q failed", dmJob.Name, dmJob.Spec.ClusterName, dmJob.Namespace)
	}
	return nil
}

func (a *regionalResourceAPI) UpdateDMJob(ctx context.Context, dmJob *v1alpha1.DMJob) (err error) {

	ctx, span := Tracer.Start(ctx, "RegionalAPI-UpdateDMJob")
	defer func() { tracing.EndSpan(span, err) }()

	span.SetAttributes(attribute.String("dm-job-name", dmJob.Name), attribute.String("namespace", dmJob.Namespace))

	err = a.regionalAPI.UpdateDMJob(ctx, dmJob)
	if err != nil {
		return errors.Annotatef(err, "update dmJob CR %q for cluster %q in namespace %q failed", dmJob.Name, dmJob.Spec.ClusterName, dmJob.Namespace)
	}
	return nil
}

func (a *regionalResourceAPI) GetDMJob(ctx context.Context, projectID, dmJobID uint64) (*v1alpha1.DMJob, error) {
	var (
		ns   = a.GetObjectNamespace(projectID)
		name = getDMJobName(dmJobID)
	)

	dmJob, err := a.regionalAPI.GetDMJob(ctx, ns, name)
	if err != nil {
		return nil, errors.Annotatef(err, "get dmJob CR %q in namespace %q failed", name, ns)
	}

	if !dmJob.DeletionTimestamp.IsZero() {
		return dmJob, errors.Errorf("dmJob %q has been marked deletion", name)
	}

	return dmJob, nil
}

func (a *regionalResourceAPI) GetObjectNamespace(projectID uint64) string {
	return dmGetObjectNamespace(projectID)
}

func (a *regionalResourceAPI) IsNotFound(err error) bool {
	return regionalapi.TCIDMIsNotFound(err)
}