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

import (
	"context"
	"strconv"
	"time"

	"github.com/pingcap/errors"
	corepb "github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"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"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
)

const (
	impOperationTimeout = 5 * time.Second
	impCheckInterval    = 500 * time.Millisecond
)

type ResourceAPI interface {
	NewImportAPIObject(importID uint64, spec *cloudv1.ImportSpec, cluster *corepb.Cluster) (*cloudv1.Import, error)
	DeleteImport(ctx context.Context, projectID, importID uint64) (err error)
	CreateImportIgnoreExists(ctx context.Context, imp *cloudv1.Import) error
	GetImport(ctx context.Context, projectID, importID uint64) (*cloudv1.Import, error)
	GetObjectNamespace(projectID uint64) string
	IsNotFound(error) bool
}

// newImportAPIObject creates a new Import CR object.
func newImportAPIObject(
	importID uint64,
	spec *cloudv1.ImportSpec,
	cluster *corepb.Cluster,
) (*cloudv1.Import, error) {
	if cluster == nil {
		return nil, errors.New("cluster is nil")
	}
	objectName := infraapi.ImportName(importID)
	labels := map[string]string{
		infraapi.TenantLabel:      strconv.FormatUint(cluster.GetTenantId(), 10),
		infraapi.ProjectLabel:     strconv.FormatUint(cluster.GetProjectId(), 10),
		infraapi.ClusterLabel:     strconv.FormatUint(cluster.GetId(), 10),
		cloudv1.ProviderTypeLabel: cluster.GetInfraProviderType(),
		cloudv1.RegionLabel:       cluster.GetRegionName(),
	}
	annotations := map[string]string{
		"import.tidbcloud.com/network-name":      cluster.GetDataPlaneInfo().GetNetworkName(),
		"import.tidbcloud.com/cluster-namespace": cluster.GetDataPlaneInfo().GetK8SNamespace(),
	}
	ns := infraapi.GetObjectNamespace(cluster.GetProjectId())
	return &cloudv1.Import{
		ObjectMeta: metav1.ObjectMeta{
			Namespace:   ns,
			Name:        objectName,
			Labels:      labels,
			Annotations: annotations,
		},
		Spec: *spec,
	}, nil
}

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) NewImportAPIObject(importID uint64, spec *cloudv1.ImportSpec, cluster *corepb.Cluster) (*cloudv1.Import, error) {
	return newImportAPIObject(importID, spec, cluster)
}

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

	ns := infraapi.GetObjectNamespace(projectID)
	importName := infraapi.ImportName(importID)
	span.SetAttributes(attribute.String("import_name", importName),
		attribute.String("namespace", ns),
	)

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

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

func (a *regionalResourceAPI) CreateImportIgnoreExists(ctx context.Context, imp *cloudv1.Import) (err error) {

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

	span.SetAttributes(attribute.String("import_name", imp.Name),
		attribute.String("namespace", imp.Namespace),
	)

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

func (a *regionalResourceAPI) GetImport(ctx context.Context, projectID, importID uint64) (*cloudv1.Import, error) {
	var (
		ns         = infraapi.GetObjectNamespace(projectID)
		importName = infraapi.ImportName(importID)
	)

	imp, err := a.regionalAPI.GetImport(ctx, ns, importName)
	if err != nil {
		return nil, errors.Annotatef(err, "get import CR %q in namespace %q failed", importName, ns)
	}

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

	return imp, nil
}

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

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