package storage

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"strings"

	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/alicloud"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/aws"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/azure"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/gcp"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/types"
	"github.com/tidbcloud/observability/vmbackupmanager/pkg/config"
	"go.uber.org/zap"
)

const (
	envStorageAcctName                = "AZURE_STORAGE_ACCOUNT_NAME"
	envStorageAccKey                  = "AZURE_STORAGE_ACCOUNT_KEY"
	envStorageAccCs                   = "AZURE_STORAGE_ACCOUNT_CONNECTION_STRING"
	azureStorageErrorCodeBlobNotFound = "BlobNotFound"
)

// ObjStorage implements the Interface for S3-compatible object storage
type ObjStorage struct {
	s3Client    providers.S3Client
	storageInfo StorageInfo
}

var _ Interface = &ObjStorage{}

// NewS3Storage creates a new S3-compatible storage implementation
func NewS3Storage(cfg *config.Config) (*ObjStorage, error) {
	storageInfo := ParseStoragePath(cfg.Destination)
	log.Info("Creating cloud storage client", zap.String("type", string(storageInfo.Type)), zap.String("bucket", storageInfo.Bucket))

	s3Cli, err := getS3Client(storageInfo.Type, cfg)
	if err != nil {
		return nil, err
	}

	return &ObjStorage{s3Cli, storageInfo}, nil
}

// DeleteFile deletes the specified object from cloud storage
func (s *ObjStorage) DeleteFile(ctx context.Context, filePath string) error {
	objectPath := s.ensurePrefix(filePath)
	return s.s3Client.Delete(ctx, s.storageInfo.Bucket, objectPath)
}

// DeleteDir deletes all files with the given prefix path.
func (s *ObjStorage) DeleteDir(ctx context.Context, prefix string) error {
	objectPath := s.ensurePrefix(prefix)
	objects, err := s.ListFiles(ctx, objectPath)
	if err != nil {
		return err
	}
	return s.s3Client.Delete(ctx, s.storageInfo.Bucket, objects...)
}

// CreateFile creates a new object with the provided data in cloud storage
func (s *ObjStorage) CreateFile(ctx context.Context, filePath string, data []byte) error {
	objectPath := s.ensurePrefix(filePath)
	reader := bytes.NewReader(data)
	log.Debug("Creating file in cloud storage", zap.String("bucket", s.storageInfo.Bucket), zap.String("path", objectPath))
	return s.s3Client.Put(ctx, s.storageInfo.Bucket, objectPath, reader)
}

// Exists checks if objects with the given filePath exists in cloud storage
func (s *ObjStorage) Exists(ctx context.Context, filePath string) (bool, error) {
	objectPath := s.ensurePrefix(filePath)
	maxKeys := int32(1)
	objects, err := s.s3Client.ListObjects(ctx, s.storageInfo.Bucket, objectPath, types.ListOptions{MaxKeys: &maxKeys})
	if err != nil {
		return false, err
	}
	return len(objects) > 0, nil
}

// ReadFile reads the contents of an object from cloud storage
func (s *ObjStorage) ReadFile(ctx context.Context, filePath string) ([]byte, error) {
	objectPath := s.ensurePrefix(filePath)
	log.Debug("Reading file from cloud storage", zap.String("bucket", s.storageInfo.Bucket), zap.String("path", objectPath))

	reader, err := s.s3Client.Get(ctx, s.storageInfo.Bucket, objectPath)
	if err != nil {
		return nil, fmt.Errorf("failed to get object: %w", err)
	}
	defer (*reader).Close()

	data, err := io.ReadAll(*reader)
	if err != nil {
		return nil, fmt.Errorf("failed to read object data: %w", err)
	}

	return data, nil
}

// ListFiles lists all objects with the given prefix from cloud storage
func (s *ObjStorage) ListFiles(ctx context.Context, prefix string) ([]string, error) {
	objectPath := s.ensurePrefix(prefix)
	log.Debug("Listing files in cloud storage", zap.String("bucket", s.storageInfo.Bucket), zap.String("prefix", objectPath))
	return s.s3Client.ListObjects(ctx, s.storageInfo.Bucket, objectPath, types.ListOptions{})
}

func getS3Client(storageType StorageType, conf *config.Config) (providers.S3Client, error) {
	var cli providers.S3Client
	switch storageType {
	case StorageTypeS3:
		log.Info("Initializing AWS S3 client")
		if conf.Vendor == constants.AWS {
			cfg, err := awsconfig.LoadDefaultConfig(context.TODO())
			if err != nil {
				return nil, errors.New("unable to get aws config")
			}
			cfg.Region = conf.Region
			cli = aws.NewS3(cfg)
		} else if conf.Vendor == constants.AliCloud && conf.CustomS3Endpoint != "" {
			log.Info("Using custom alicloud endpoint", zap.String("endpoint", conf.CustomS3Endpoint))

			// Get access key and secret key from environment variables
			accessKeyID := os.Getenv("AWS_ACCESS_KEY_ID")
			secretAccessKey := os.Getenv("AWS_SECRET_ACCESS_KEY")

			if accessKeyID == "" || secretAccessKey == "" {
				return nil, fmt.Errorf("AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables must be set for AliCloud")
			}

			log.Info("Using AliCloud with access key authentication")
			provider := credentials.CredentialsProviderFunc(func(ctx context.Context) (credentials.Credentials, error) {
				return credentials.Credentials{
					AccessKeyID:     accessKeyID,
					AccessKeySecret: secretAccessKey,
				}, nil
			})
			cfg := oss.LoadDefaultConfig().WithRegion(conf.Region).WithCredentialsProvider(provider)
			// Use default ZRS for backup manager
			cli = alicloud.NewS3(cfg, "")
		}

	case StorageTypeGCS:
		log.Info("Initializing Google Cloud Storage client")
		cli = gcp.NewCloudStorage("", "") // we don't need to set project and region for gcs client for operations on vmBackupmanager
	case StorageTypeAzure:
		log.Info("Initializing Azure Blob Storage client")
		cred, err := azidentity.NewDefaultAzureCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create default Azure credentials: %w", err)
		}
		// load storage account from environment variable
		sc := os.Getenv(envStorageAcctName)
		if sc == "" {
			return nil, fmt.Errorf("azure storage account is not set")
		}
		if azCli, err := azure.NewBlobStorageClient("", "", sc, cred); err != nil {
			return nil, err
		} else {
			cli = azCli
		}
	default:
		return nil, fmt.Errorf("not supported provider")
	}
	return cli, nil
}

// GetFullPath returns the full path for the given filePath.
func (s *ObjStorage) GetFullPath(filePath string) string {
	return s.storageInfo.FullPath + filePath
}

func (s *ObjStorage) ensurePrefix(filePath string) string {
	if s.storageInfo.Prefix == "" || strings.HasPrefix(filePath, s.storageInfo.Prefix) {
		return filePath
	}
	return s.storageInfo.Prefix + filePath
}