package impl

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"os"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/storage/armstorage"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/sas"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/pkgv2/bizerr"
)

const (
	containerPolicyName = "br"
)

type azureBucketManagerImpl struct {
	resourceGroup string
}

func NewAzureBucketManagerModel(resourceGroup string) bucket.AzureBucketManagerModel {
	return &azureBucketManagerImpl{
		resourceGroup: resourceGroup,
	}
}

// CreateBucket creates a Storage Account and a Container
func (m *azureBucketManagerImpl) CreateBucket(ctx context.Context, req *domain.CreateAzureBucketReq) bizerr.BizError {
	factory, err := getClientFactory(req.SubscriptionID)
	if err != nil {
		return bizerr.WrapError(err, "get client factory")
	}
	tags := make(map[string]*string, len(req.Tags))
	for k, v := range req.Tags {
		tags[k] = to.Ptr(v)
	}
	params := armstorage.AccountCreateParameters{
		Kind:     to.Ptr(armstorage.KindStorageV2),
		Location: to.Ptr(req.Region),
		SKU: &armstorage.SKU{
			Name: to.Ptr(armstorage.SKUNameStandardLRS),
		},
		Properties: &armstorage.AccountPropertiesCreateParameters{
			EnableHTTPSTrafficOnly: to.Ptr(true),
			AllowBlobPublicAccess:  to.Ptr(false),
			Encryption: &armstorage.Encryption{
				Services: &armstorage.EncryptionServices{
					Blob: &armstorage.EncryptionService{
						Enabled: to.Ptr(true),
					},
				},
				KeySource: to.Ptr(armstorage.KeySourceMicrosoftStorage),
			},
		},
		Tags: tags,
	}
	accountName := genStorageAccountName(req.Name)
	poller, err := factory.NewAccountsClient().BeginCreate(ctx, m.resourceGroup, accountName, params, nil)
	if err != nil {
		return bizerr.WrapError(err, "create storage account")
	}
	_, err = poller.PollUntilDone(ctx, nil)
	if err != nil {
		return bizerr.WrapError(err, "poll creating status")
	}
	return nil
}

func (m *azureBucketManagerImpl) CheckBucketIfExists(
	ctx context.Context,
	req *domain.CheckAzureBucketReq,
) bizerr.BizError {
	factory, err := getClientFactory(req.SubscriptionID)
	if err != nil {
		return bizerr.WrapError(err, "get client factory")
	}
	accountName := genStorageAccountName(req.Name)
	_, err = factory.NewAccountsClient().GetProperties(ctx, m.resourceGroup, accountName, nil)
	if err != nil {
		return bizerr.WrapError(err, "get bucket")
	}
	return nil
}

func (m *azureBucketManagerImpl) CreateContainer(
	ctx context.Context,
	req *domain.CreateAzureBucketContainerReq,
) bizerr.BizError {
	factory, err := getClientFactory(req.SubscriptionID)
	if err != nil {
		return bizerr.WrapError(err, "get client factory")
	}
	cli := factory.NewBlobContainersClient()
	container := armstorage.BlobContainer{
		ContainerProperties: &armstorage.ContainerProperties{
			PublicAccess: to.Ptr(armstorage.PublicAccessNone),
		},
	}
	accountName := genStorageAccountName(req.Name)
	// container name does not have limitation as storage account
	// so we use bucket name as container name for convenience
	containerName := req.Name
	_, err = cli.Create(ctx, m.resourceGroup, accountName, containerName, container, nil)
	if err != nil {
		return bizerr.WrapError(err, "create blob container")
	}
	return nil
}

// EnableBucketEncryption implements bucket.AzureBucketManagerModel.
func (m *azureBucketManagerImpl) EnableBucketEncryption(
	ctx context.Context, req *domain.EnableAzureBucketEncryptionReq,
) bizerr.BizError {
	factory, err := getClientFactory(req.SubscriptionID)
	if err != nil {
		return bizerr.WrapError(err, "get client factory")
	}

	accountName := genStorageAccountName(req.Name)
	params := armstorage.AccountUpdateParameters{
		Identity: &armstorage.Identity{
			Type: to.Ptr(armstorage.IdentityTypeUserAssigned),
			UserAssignedIdentities: map[string]*armstorage.UserAssignedIdentity{
				req.ManagedIdentityResourceID: {},
			},
		},
		Properties: &armstorage.AccountPropertiesUpdateParameters{
			Encryption: &armstorage.Encryption{
				EncryptionIdentity: &armstorage.EncryptionIdentity{
					EncryptionFederatedIdentityClientID: to.Ptr(req.ApplicationID),
					EncryptionUserAssignedIdentity:      to.Ptr(req.ManagedIdentityResourceID),
				},
				KeySource: to.Ptr(armstorage.KeySourceMicrosoftKeyvault),
				KeyVaultProperties: &armstorage.KeyVaultProperties{
					KeyName:     &req.KeyName,
					KeyVaultURI: pointer.To(buildKeyVaultURI(req.KeyVaultName)),
					KeyVersion:  nil, // is not required
				},
				Services: &armstorage.EncryptionServices{
					Blob: &armstorage.EncryptionService{
						Enabled: to.Ptr(true),
					},
					File: &armstorage.EncryptionService{
						Enabled: to.Ptr(true),
					},
					Queue: &armstorage.EncryptionService{},
					Table: &armstorage.EncryptionService{},
				},
			},
		},
	}
	_, err = factory.NewAccountsClient().Update(ctx, m.resourceGroup, accountName, params, nil)
	if err != nil {
		return bizerr.WrapError(err, "get bucket")
	}
	return nil
}

func buildKeyVaultURI(vaultName string) string {
	return fmt.Sprintf("https://%s.vault.azure.net/", vaultName)
}

// GenerateSASToken generates an SAS token for the specified storage account
// Log backup needs a long-term SAS token, so we use shared access key to generate SAS token
// BR will get container properties to check if container exist
// This API requires Account level SAS token
// https://learn.microsoft.com/en-us/rest/api/storageservices/get-container-properties?tabs=shared-access-signatures#authorization
// BR should avoid checking container existence to allow users using service SAS
// As service SAS has minimal permissions and better revokation options
func (m *azureBucketManagerImpl) GenerateSASToken(
	ctx context.Context,
	req *domain.GenerateAzureSASTokenReq,
) (string, bizerr.BizError) {
	factory, err := getClientFactory(req.SubscriptionID)
	if err != nil {
		return "", bizerr.WrapError(err, "get client factory")
	}
	accountsClient := factory.NewAccountsClient()
	accountName := genStorageAccountName(req.Name)
	resp, err := accountsClient.ListKeys(ctx, m.resourceGroup, accountName, nil)
	if err != nil {
		return "", bizerr.WrapError(err, "list keys")
	}
	if len(resp.Keys) == 0 {
		return "", bizerr.WrapError(err, "no keys found")
	}
	// There are two keys for a storage account
	// The first key is used in normal case
	// The second key is used for rotation
	key := resp.Keys[0]
	cred, err := service.NewSharedKeyCredential(accountName, *key.Value)
	if err != nil {
		return "", bizerr.WrapError(err, "new shared key credential")
	}
	// TODO: fine-tune the permissions for BR
	permissions := sas.AccountPermissions{
		Read:   true,
		List:   true,
		Add:    true,
		Create: true,
		Write:  true,
		Delete: true,
	}
	resourceTypes := sas.AccountResourceTypes{
		Container: true,
		Object:    true,
	}
	now := time.Now().UTC()
	params := sas.AccountSignatureValues{
		Protocol:      sas.ProtocolHTTPS, // always use https
		Permissions:   permissions.String(),
		StartTime:     now,
		ExpiryTime:    now.Add(req.Expiration),
		ResourceTypes: resourceTypes.String(),
	}

	sas, err := params.SignWithSharedKey(cred)
	if err != nil {
		return "", bizerr.WrapError(err, "generate sas token")
	}
	return sas.Encode(), nil
}

func (m *azureBucketManagerImpl) DeleteBucketIgnoreNotFound(
	ctx context.Context, req *domain.DeleteAzureBucketReq,
) bizerr.BizError {
	factory, err := getClientFactory(req.SubscriptionID)
	if err != nil {
		return bizerr.WrapError(err, "get client factory")
	}
	accountName := genStorageAccountName(req.Name)
	_, err = factory.NewAccountsClient().Delete(ctx, m.resourceGroup, accountName, nil)
	if err != nil {
		return bizerr.WrapError(err, "delete storage account")
	}
	return nil
}

func getClientFactory(subscriptionID string) (*armstorage.ClientFactory, error) {
	cred, err := getCredential()
	if err != nil {
		return nil, err
	}
	return armstorage.NewClientFactory(subscriptionID, cred, nil)
}

func getCredential() (azcore.TokenCredential, error) {
	tokenFile := os.Getenv("AZURE_FEDERATED_TOKEN_FILE")
	if tokenFile == "" { // no federated token, use default credential
		return azidentity.NewDefaultAzureCredential(nil)
	}
	opts := &azidentity.WorkloadIdentityCredentialOptions{
		ClientID:      os.Getenv("AZURE_CLIENT_ID"),
		TenantID:      os.Getenv("AZURE_TENANT_ID"),
		TokenFilePath: tokenFile,
	}
	return azidentity.NewWorkloadIdentityCredential(opts)
}

// genStorageAccountName gets a storage account name from bucket name
// Because Azure Stroage Account name has following limitations:
// - 3 and 24 characters in length
// - use numbers and lower-case only
func genStorageAccountName(name string) string {
	hash := md5.Sum([]byte(name))
	return hex.EncodeToString(hash[:])[:24]
}