package provider

import (
	"fmt"
	"net/url"
	"regexp"
	"strings"

	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"k8s.io/utils/ptr"
)

type BRStorageProvider interface {
	GetProviderName() string
	// OverrideBackup method override Backup CR with cloud vendor specific config
	OverrideBackup(*opv1alpha1.Backup)
	// OverrideRestore method override Restore CR with cloud vendor specific config
	OverrideRestore(*opv1alpha1.Restore)
}

func NewBRStorageProvider(storageUri string, region string, secret string) (BRStorageProvider, error) {
	u, err := url.Parse(storageUri)
	if err != nil {
		return nil, err
	}
	switch u.Scheme {
	case "azblob":
		return newAzureBRStorage(u)
	case "s3":
		return nil, fmt.Errorf("s3 provider unimplemented")
	case "gcs":
		return nil, fmt.Errorf("gcs provider unimplemented")
	default:
		// unknown provider
		return nil, fmt.Errorf("unknown storage provider")
	}
}

type azureBRStorage struct {
	providerName   string
	storageAccount string
	container      string
	prefix         string
	sasToken       string
}

func newAzureBRStorage(uri *url.URL) (BRStorageProvider, error) {
	re := regexp.MustCompile(`^([a-z0-9]+)\.blob\.core\.windows\.net`)
	match := re.FindStringSubmatch(uri.Host)
	if len(match) <= 1 {
		return nil, fmt.Errorf("invalid bucket uri for azure blob storage, no storage account specified")
	}
	account := match[1]
	s := strings.Split(uri.Path, "/")
	if len(s) < 2 {
		return nil, fmt.Errorf("invalid bucket uri for azure blob storage, no container specified")
	}
	container := s[1]
	if container == "" {
		return nil, fmt.Errorf("invalid bucket uri for azure blob storage, no container specified")
	}
	directory := ""
	if len(s) > 2 {
		directory = "/" + strings.Join(s[2:], "/")
	}
	sasToken := uri.RawQuery
	if sasToken == "" {
		return nil, fmt.Errorf("invalid bucket uri for azure blob storage, no SAS token specified")
	}
	return &azureBRStorage{
		providerName:   "Azure",
		storageAccount: account,
		container:      container,
		prefix:         directory,
		sasToken:       sasToken,
	}, nil
}

func (p *azureBRStorage) GetProviderName() string {
	return p.providerName
}

func (p *azureBRStorage) OverrideBackup(backup *opv1alpha1.Backup) {
	backup.Spec.StorageProvider = opv1alpha1.StorageProvider{
		Azblob: &opv1alpha1.AzblobStorageProvider{
			StorageAccount: p.storageAccount,
			Container:      p.container,
			SasToken:       p.sasToken,
			Prefix:         p.prefix,
		},
	}
	backup.Spec.BR.SendCredToTikv = ptr.To(true)
}

func (p *azureBRStorage) OverrideRestore(restore *opv1alpha1.Restore) {
	restore.Spec.StorageProvider = opv1alpha1.StorageProvider{
		Azblob: &opv1alpha1.AzblobStorageProvider{
			StorageAccount: p.storageAccount,
			Container:      p.container,
			SasToken:       p.sasToken,
			Prefix:         p.prefix,
		},
	}
	restore.Spec.BR.SendCredToTikv = ptr.To(true)
}