package infraapi

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/pingcap/errors"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/tracing"
	"go.opentelemetry.io/otel/attribute"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
)

const (
	impDeleteTimeout = 5 * time.Second
	impCheckInterval = 50 * time.Millisecond
)

func (a *API) GetImport(ctx context.Context, projectID, importID uint64) (*cloudv1.Import, error) {
	var (
		imp        cloudv1.Import
		ns         = Namespace(a.projectName, projectID)
		importName = ImportName(importID)
	)
	err := a.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: importName}, &imp)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil, err
		}
		return nil, errors.Annotatef(err, "get import CR %q in namespace %q failed", importName, ns)
	}

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

	return &imp, nil
}

func (a *API) DeleteImport(ctx context.Context, projectID, importID uint64) (err error) {
	ctx, span := Tracer.Start(ctx, "API.DeleteImport")
	defer func() { tracing.EndSpan(span, err) }()
	importName := ImportName(importID)
	span.SetAttributes(attribute.String("import_name", importName),
		attribute.String("project_id", fmt.Sprintf("%v", projectID)),
	)
	imp := &cloudv1.Import{ObjectMeta: metav1.ObjectMeta{
		Namespace: Namespace(a.projectName, projectID),
		Name:      importName,
	}}
	err = a.client.Delete(ctx, imp)
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Annotatef(err, "delete import CR %q in namespace %q failed", importName, imp.Namespace)
	}

	return wait.PollImmediate(impCheckInterval, impDeleteTimeout, func() (bool, error) {
		imp, err := a.GetImport(context.TODO(), projectID, importID)
		if err != nil && !apierrors.IsNotFound(err) {
			return false, err
		}
		return imp == nil, nil
	})
}

func (a *API) CreateAwsImportIgnoreExists(
	ctx context.Context,
	tenantID, projectID, clusterID, importID uint64,
	userName, passWord, sourceUrl, region, filter, storageRequest string,
	dataFormat cloudv1.DataFormat,
	csvFormat *cloudv1.CsvFormat,
	roleARN,
	objectNamePattern, targetTablePattern string,
) error {
	var filters []string
	imp := &cloudv1.Import{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: Namespace(a.projectName, projectID),
			Name:      ImportName(importID),
			Labels: map[string]string{
				tenantLabel:  strconv.FormatUint(tenantID, 10),
				projectLabel: strconv.FormatUint(projectID, 10),
				clusterLabel: strconv.FormatUint(clusterID, 10),
			},
		},
		Spec: cloudv1.ImportSpec{
			ClusterName: strconv.FormatUint(clusterID, 10),
			Source: cloudv1.Source{
				UserName: userName,
				Password: passWord,
				Format:   dataFormat,
				Storage: cloudv1.Storage{
					S3: &cloudv1.S3{
						URL:    sourceUrl,
						Region: region,
					},
				},
				Filters:            append(filters, filter),
				CsvFormat:          csvFormat,
				ObjectNamePattern:  objectNamePattern,
				TargetTablePattern: targetTablePattern,
			},
			AccessConfig: cloudv1.AccessConfig{
				RoleArn: roleARN,
			},
			StorageRequest: storageRequest,
		},
	}
	err := a.client.Create(ctx, imp)
	if err != nil && !apierrors.IsAlreadyExists(err) {
		return errors.Annotatef(err, "create aws import CR %q for cluster %q in namespace %q failed", imp.Name, imp.Spec.ClusterName, imp.Namespace)
	}
	return nil
}

func (a *API) CreateGcpImportIgnoreExists(
	ctx context.Context,
	tenantID, projectID, clusterID, importID uint64,
	userName, passWord, sourceUrl, region, filter, storageRequest string,
	dataFormat cloudv1.DataFormat, csvFormat *cloudv1.CsvFormat,
	objectNamePattern, targetTablePattern string,
) error {
	var filters []string
	imp := &cloudv1.Import{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: Namespace(a.projectName, projectID),
			Name:      ImportName(importID),
			Labels: map[string]string{
				tenantLabel:  strconv.FormatUint(tenantID, 10),
				projectLabel: strconv.FormatUint(projectID, 10),
				clusterLabel: strconv.FormatUint(clusterID, 10),
			},
		},
		Spec: cloudv1.ImportSpec{
			ClusterName: strconv.FormatUint(clusterID, 10),
			Source: cloudv1.Source{
				UserName: userName,
				Password: passWord,
				Format:   dataFormat,
				Storage: cloudv1.Storage{
					GCS: &cloudv1.GCS{
						URL:    sourceUrl,
						Region: region,
					},
				},
				Filters:            append(filters, filter),
				CsvFormat:          csvFormat,
				ObjectNamePattern:  objectNamePattern,
				TargetTablePattern: targetTablePattern,
			},
			StorageRequest: storageRequest,
		},
	}
	err := a.client.Create(ctx, imp)
	if err != nil && !apierrors.IsAlreadyExists(err) {
		return errors.Annotatef(err, "create gcp import CR %q for cluster %q in namespace %q failed", imp.Name, imp.Spec.ClusterName, imp.Namespace)
	}
	return nil
}

func ImportName(importID uint64) string {
	return fmt.Sprintf("import-%d", importID)
}