package bucket

import (
	"context"
	"fmt"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task/ack"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task/aks"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task/eks"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type BucketQueryer interface {
	GetDataplanePermissionStatus(ctx context.Context, namespace, bucketName, dpid string) (*task.BucketDataplanePermissionStatus, error)
	GetLifecycleRulesStatus(ctx context.Context, bucket *v1alpha1.Bucket) (*task.BucketLifecycleRulesStatus, error)
	GetBucketAccessPath(ctx context.Context, bucket *v1alpha1.Bucket) (string, error)
}

type bucketQueryer struct {
	cli client.Client
}

func NewBucketQueryer(cli client.Client) BucketQueryer {
	return &bucketQueryer{cli: cli}
}

func (q *bucketQueryer) getVendorQueryer(bucket *v1alpha1.Bucket) task.BucketVendorQueryer {
	provider := bucket.Labels["tags.tidbcloud.com/cloud-provider"]
	switch provider {
	case consts.ProviderTypeACK:
		return ack.NewAliQueryer(q.cli)
	case consts.ProviderTypeAKS:
		return aks.NewAzureQueryer(q.cli)
	case consts.ProviderTypeEKS:
		return eks.NewAWSQueryer(q.cli)
	default: // Notice: aws is the first provider, same early created buckets didn'have the label
		return eks.NewAWSQueryer(q.cli)
	}
}

func (q *bucketQueryer) GetDataplanePermissionStatus(ctx context.Context, namespace, bucketName, dpid string) (*task.BucketDataplanePermissionStatus, error) {
	bucket := &v1alpha1.Bucket{}
	if err := q.cli.Get(ctx, types.NamespacedName{
		Namespace: namespace,
		Name:      bucketName,
	}, bucket); err != nil {
		return nil, fmt.Errorf("get bucket error: %w", err)
	}
	dpIdFound := false
	for _, grantedDataplaneId := range bucket.Spec.GrantedDataplaneIds {
		if dpid == grantedDataplaneId {
			dpIdFound = true
			break
		}
	}
	if !dpIdFound {
		return ptr.To(task.BucketDataplanePermissionStatusNotExists), nil
	}

	vendorQueryer := q.getVendorQueryer(bucket)
	permissionStatus, err := vendorQueryer.GetDataplanePermissionStatus(ctx, bucket, dpid)
	if err != nil {
		return nil, fmt.Errorf("get bucket dataplane permission status error: %w", err)
	}
	return permissionStatus, nil
}

func (q *bucketQueryer) GetLifecycleRulesStatus(ctx context.Context, bucket *v1alpha1.Bucket) (*task.BucketLifecycleRulesStatus, error) {
	// check status in vendor
	vendorQueryer := q.getVendorQueryer(bucket)
	lifecycleRulesStatus, err := vendorQueryer.GetLifecycleRulesStatus(ctx, bucket)
	if err != nil {
		return nil, fmt.Errorf("get bucket lifecycle rules status error: %w", err)
	}

	return lifecycleRulesStatus, nil
}

func (q *bucketQueryer) GetBucketAccessPath(ctx context.Context, bucket *v1alpha1.Bucket) (string, error) {
	return q.getVendorQueryer(bucket).GetBucketAccessPath(ctx, bucket)
}