package checkpoint

import (
	"context"
	"fmt"
	"os"
	"strings"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	sdkstorage "github.com/pingcap/metering_sdk/storage"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/utils/storage/aws"
	"go.uber.org/zap"
)

// Factory creates checkpoint managers based on configuration
type Factory struct {
	conf *config.ServiceConfig
	log  *zap.Logger
}

// NewFactory creates a new checkpoint factory
func NewFactory(conf *config.ServiceConfig, log *zap.Logger) *Factory {
	return &Factory{
		conf: conf,
		log:  log,
	}
}

// CreateCheckpointManager creates a checkpoint manager based on storage configuration
func (f *Factory) CreateCheckpointManager() (NextGenCheckpoint, error) {
	if f.conf.NextGenMeteringConfig == nil {
		return nil, fmt.Errorf("NextGenMeteringConfig is required")
	}

	// Use the service's vendor configuration from the top level
	vendor := f.conf.Vendor
	if vendor == "" {
		return nil, fmt.Errorf("vendor is required in service configuration")
	}

	switch strings.ToLower(vendor) {
	case constants.VendorAWS:
		return f.createS3CheckpointManager()
	case constants.VendorAlicloud:
		return f.createAlicloudCheckpointManager()
	case constants.VendorGCP:
		// TODO: Implement GCP checkpoint manager
		return nil, fmt.Errorf("GCP checkpoint manager not implemented yet")
	case constants.VendorAzure:
		return f.createAzureCheckpointManager()
	default:
		return nil, fmt.Errorf("unsupported storage vendor: %s", vendor)
	}
}

// createS3CheckpointManager creates an S3-based checkpoint manager
func (f *Factory) createS3CheckpointManager() (NextGenCheckpoint, error) {
	awsCfg, err := awsconfig.LoadDefaultConfig(context.Background())
	if err != nil {
		return nil, fmt.Errorf("failed to load AWS config: %w", err)
	}

	awsCfg.Region = f.conf.Region

	return f.createS3CheckpointManagerWithConfig(awsCfg, false)
}

func (f *Factory) createAlicloudCheckpointManager() (NextGenCheckpoint, error) {
	accessKey := strings.TrimSpace(os.Getenv(constants.AlicloudAccessKeyIDEnv))
	secretKey := strings.TrimSpace(os.Getenv(constants.AlicloudAccessKeySecretEnv))

	if accessKey == "" || secretKey == "" {
		return nil, fmt.Errorf("%s and %s environment variables are required for Alicloud", constants.AlicloudAccessKeyIDEnv, constants.AlicloudAccessKeySecretEnv)
	}

	bucketEndpoint := strings.TrimSpace(f.conf.NextGenMeteringConfig.BucketEndpoint)
	if bucketEndpoint == "" {
		return nil, fmt.Errorf("bucketEndpoint is required in NextGenMeteringConfig for Alicloud")
	}

	region := strings.TrimSpace(f.conf.Region)
	if region == "" {
		return nil, fmt.Errorf("region is required in service configuration")
	}

	awsCfg, err := awsconfig.LoadDefaultConfig(
		context.Background(),
		awsconfig.WithRegion(region),
		awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to load Alicloud AWS-compatible config: %w", err)
	}

	awsCfg.BaseEndpoint = awssdk.String(bucketEndpoint)

	return f.createS3CheckpointManagerWithConfig(awsCfg, true)
}

func (f *Factory) createAzureCheckpointManager() (NextGenCheckpoint, error) {
	bucket := strings.TrimSpace(f.conf.Bucket)
	if bucket == "" {
		return nil, fmt.Errorf("bucket is required in service configuration")
	}

	accountName := strings.TrimSpace(os.Getenv(constants.AzureStorageAccountNameEnv))
	if accountName == "" {
		return nil, fmt.Errorf("%s environment variable is required for Azure", constants.AzureStorageAccountNameEnv)
	}

	checkpointPrefix := "checkpoints"
	if prefix := strings.TrimSpace(f.conf.NextGenMeteringConfig.CheckpointPrefix); prefix != "" {
		checkpointPrefix = prefix
	}

	providerCfg := &sdkstorage.ProviderConfig{
		Type:   sdkstorage.ProviderTypeAzure,
		Bucket: bucket,
		Prefix: checkpointPrefix,
		Azure: &sdkstorage.AzureConfig{
			AccountName: accountName,
		},
	}

	if accountKey := strings.TrimSpace(os.Getenv(constants.AzureStorageAccountKeyEnv)); accountKey != "" {
		providerCfg.Azure.AccountKey = accountKey
	}
	if sasToken := strings.TrimSpace(os.Getenv(constants.AzureStorageSASTokenEnv)); sasToken != "" {
		providerCfg.Azure.SASToken = sasToken
	}

	provider, err := sdkstorage.NewObjectStorageProvider(providerCfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create Azure storage provider: %w", err)
	}

	f.log.Info("creating Azure checkpoint manager",
		zap.String("container", bucket),
		zap.String("prefix", checkpointPrefix),
		zap.Bool("has_account_key", providerCfg.Azure.AccountKey != ""),
		zap.Bool("has_sas_token", providerCfg.Azure.SASToken != ""))

	return NewObjectStorageCheckpointManager(provider, f.log), nil
}

func (f *Factory) createS3CheckpointManagerWithConfig(awsCfg awssdk.Config, hoststyle bool) (NextGenCheckpoint, error) {
	bucket := strings.TrimSpace(f.conf.Bucket)
	if bucket == "" {
		return nil, fmt.Errorf("bucket is required in service configuration")
	}

	if strings.TrimSpace(awsCfg.Region) == "" {
		return nil, fmt.Errorf("region is required in service configuration")
	}

	checkpointPrefix := "checkpoints"
	if prefix := strings.TrimSpace(f.conf.NextGenMeteringConfig.CheckpointPrefix); prefix != "" {
		checkpointPrefix = prefix
	}

	cachePath := "/tmp"
	if configuredCachePath := strings.TrimSpace(f.conf.NextGenMeteringConfig.CachePath); configuredCachePath != "" {
		cachePath = configuredCachePath
	}
	var s3Client *aws.AwsProviderS3
	if hoststyle {
		s3Client = aws.NewAwsProviderS3UseHostStyle(awsCfg, bucket)
	} else {
		s3Client = aws.NewAwsProviderS3(awsCfg, bucket)
	}

	if f.conf.NextGenMeteringConfig != nil {
		if roleArn := strings.TrimSpace(f.conf.NextGenMeteringConfig.RoleArn); roleArn != "" {
			s3Client = s3Client.AssumeRole(context.Background(), roleArn)
		}
	}

	logFields := []zap.Field{
		zap.String("bucket", bucket),
		zap.String("prefix", checkpointPrefix),
		zap.String("cache_path", cachePath),
	}
	if endpoint := strings.TrimSpace(f.conf.NextGenMeteringConfig.BucketEndpoint); endpoint != "" {
		logFields = append(logFields, zap.String("endpoint", endpoint))
	}

	f.log.Info("creating S3 checkpoint manager", logFields...)

	return NewS3CheckpointManager(s3Client, bucket, checkpointPrefix, cachePath, f.log), nil
}