package cloud_storage

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"os"
	"strings"
	"time"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	aliCredetials "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	openapicred "github.com/aliyun/credentials-go/credentials"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/aws-sdk-go-v2/service/sts"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/import/cloud_storage/creds"
	"github.com/tidbcloud/pkgv2/log"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
)

var checkAccessRange = aws.String("bytes=0-1")

type S3 struct {
	Client *s3.Client
	Bucket string
	Prefix *string
	Region string
	Schema string
}

// S3Config holds all configuration for S3 client initialization.
type S3Config struct {
	URI             string
	RoleArn         *string
	ExternalId      *string
	AccessKeyID     *string
	SecretAccessKey *string
	Region          *string
	client          *s3.Client
}

type S3Option func(*S3Config)

func WithAssumeRole(roleArn, externalId string) S3Option {
	return func(cfg *S3Config) {
		cfg.RoleArn = &roleArn
		cfg.ExternalId = &externalId
	}
}

func WithAccessKey(accessKeyID, secretAccessKey string) S3Option {
	return func(cfg *S3Config) {
		cfg.AccessKeyID = &accessKeyID
		cfg.SecretAccessKey = &secretAccessKey
	}
}

func WithS3Client(client *s3.Client) S3Option {
	return func(cfg *S3Config) {
		cfg.client = client
	}
}

func WithS3Region(region string) S3Option {
	return func(cfg *S3Config) {
		cfg.Region = &region
	}
}

// NewS3 creates a new S3 CloudStorage instance using functional options.
func NewS3(ctx context.Context, uri string, opts ...S3Option) (CloudStorage, bizerr.BizError) {
	cfg := &S3Config{URI: uri}
	for _, opt := range opts {
		opt(cfg)
	}

	schema, bucket, prefix, bizError := parseS3Params(cfg.URI)
	if bizError != nil {
		return nil, bizError
	}

	awsCfg, err := config.LoadDefaultConfig(ctx,
		config.WithRegion("us-west-2"), // temporary, will be overridden below
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "LoadDefaultConfig failed")
	}

	var region string
	if cfg.Region != nil {
		region = *cfg.Region
	} else {
		region, err = manager.GetBucketRegion(ctx, s3.NewFromConfig(awsCfg), bucket)
		if err != nil {
			log.Error(ctx, "failed to get bucket region", zap.Error(err), zap.String("bucket", bucket))
			var bnf manager.BucketNotFound
			if errors.As(err, &bnf) {
				return nil, bizerr.New(ErrCodeS3BucketNotFound, errMsgBucketNotFound)
			}
			return nil, bizerr.WrapError(err, "GetBucketRegion failed")
		}
	}
	awsCfg.Region = region

	if cfg.client != nil {
		return &S3{
			Client: cfg.client,
			Bucket: bucket,
			Prefix: prefix,
			Region: region,
			Schema: schema,
		}, nil
	}

	if cfg.RoleArn != nil && cfg.ExternalId != nil {
		creds := stscreds.NewAssumeRoleProvider(sts.NewFromConfig(awsCfg), *cfg.RoleArn,
			func(o *stscreds.AssumeRoleOptions) {
				o.ExternalID = cfg.ExternalId
			})
		awsCfg.Credentials = aws.NewCredentialsCache(creds)
	} else if cfg.AccessKeyID != nil && cfg.SecretAccessKey != nil {
		if len(*cfg.AccessKeyID) != 20 && !strings.HasPrefix(*cfg.SecretAccessKey, "AKIA") {
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid access key id")
		}
		awsCfg.Credentials = aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider(*cfg.AccessKeyID, *cfg.SecretAccessKey, ""))
	} else {
		return nil, bizerr.New(bizerr.ErrCodeParams, "S3Config: must provide either (accessKeyID & secretAccessKey) or (roleArn & externalId)")
	}

	return &S3{
		Client: s3.NewFromConfig(awsCfg),
		Bucket: bucket,
		Prefix: prefix,
		Region: region,
		Schema: schema,
	}, nil
}

type S3CompatibleConfig struct {
	URI             string
	AccessKeyID     *string
	SecretAccessKey *string
	Endpoint        *string
	Region          *string
}

type S3CompatibleOption func(*S3CompatibleConfig)

func WithS3CompatibleAccessKey(accessKeyID, secretAccessKey string) S3CompatibleOption {
	return func(cfg *S3CompatibleConfig) {
		cfg.AccessKeyID = &accessKeyID
		cfg.SecretAccessKey = &secretAccessKey
	}
}

func WithS3CompatibleEndpoint(endpoint string) S3CompatibleOption {
	return func(cfg *S3CompatibleConfig) {
		cfg.Endpoint = &endpoint
	}
}

func WithS3CompatibleRegion(region string) S3CompatibleOption {
	return func(cfg *S3CompatibleConfig) {
		cfg.Region = &region
	}
}

// NewS3Compatible creates a new S3-compatible CloudStorage instance using functional options.
func NewS3Compatible(ctx context.Context, uri string, opts ...S3CompatibleOption) (CloudStorage, bizerr.BizError) {
	cfg := &S3CompatibleConfig{URI: uri}
	for _, opt := range opts {
		opt(cfg)
	}

	schema, bucket, prefix, bizError := parseS3Params(cfg.URI)
	if bizError != nil {
		return nil, bizError
	}

	if cfg.AccessKeyID == nil || cfg.SecretAccessKey == nil || cfg.Endpoint == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "S3CompatibleConfig: must provide accessKeyID, secretAccessKey, and endpoint")
	}

	awsCfg, err := config.LoadDefaultConfig(ctx,
		// Cloudflare requires the region to be set to "auto"
		config.WithRegion("auto"),
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "LoadDefaultConfig failed")
	}

	// Cloudflare requires an authorized access key to retrieve the bucket region.
	awsCfg.Credentials = credentials.NewStaticCredentialsProvider(*cfg.AccessKeyID, *cfg.SecretAccessKey, "")
	client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
		o.BaseEndpoint = cfg.Endpoint
	})

	var region string
	if cfg.Region != nil {
		region = *cfg.Region
	} else {
		region, err = manager.GetBucketRegion(ctx, client, bucket)
		if err != nil {
			log.Error(ctx, "failed to get bucket region", zap.Error(err), zap.String("bucket", bucket))
			var bnf manager.BucketNotFound
			if errors.As(err, &bnf) {
				return nil, bizerr.New(ErrCodeS3BucketNotFound, errMsgBucketNotFound)
			}
			return nil, bizerr.WrapError(err, "GetBucketRegion failed")
		}
	}

	awsCfg.Region = region

	s3Client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
		o.BaseEndpoint = cfg.Endpoint
	})

	return &S3{
		Schema: schema,
		Client: s3Client,
		Bucket: bucket,
		Prefix: prefix,
		Region: region,
	}, nil
}

type OSSConfig struct {
	URI                 string
	AccessKeyID         *string
	AccessKeySecret     *string
	RoleArn             *string
	ExternalId          *string
	UseInternalEndpoint *bool // Use internal endpoint for OSS
	Duration            time.Duration
	RefreshWindow       time.Duration
	Region              *string
}

type OSSOption func(*OSSConfig)

func WithOSSAccessKey(accessKeyID, accessKeySecret string) OSSOption {
	return func(cfg *OSSConfig) {
		cfg.AccessKeyID = &accessKeyID
		cfg.AccessKeySecret = &accessKeySecret
	}
}

func WithOSSAssumeRole(roleArn, externalId string) OSSOption {
	return func(cfg *OSSConfig) {
		cfg.RoleArn = &roleArn
		cfg.ExternalId = &externalId
	}
}

func WithOSSRegion(region string) OSSOption {
	return func(cfg *OSSConfig) {
		cfg.Region = &region
	}
}

func UseInternalEndpoint() OSSOption {
	return func(cfg *OSSConfig) {
		cfg.UseInternalEndpoint = aws.Bool(true)
	}
}

// WithRoleDuration sets the duration and refresh window for AssumeRole credentials.
// Duration is the total validity period of the role, and RefreshWindow is the time before expiration to refresh credentials.
func WithRoleDuration(duration time.Duration, refreshWindow time.Duration) OSSOption {
	return func(cfg *OSSConfig) {
		cfg.Duration = duration
		cfg.RefreshWindow = refreshWindow
	}
}

func GetOssBucketRegion(ctx context.Context, uri string, opts ...OSSOption) (string, bizerr.BizError) {
	cfg := &OSSConfig{URI: uri}
	for _, opt := range opts {
		opt(cfg)
	}
	_, bucket, _, bizError := parseS3Params(cfg.URI)
	if bizError != nil {
		return "", bizError
	}

	provider, err := buildAliCredentialsProvider(cfg)
	if err != nil {
		return "", bizerr.WrapError(err, "failed to build credentials provider")
	}

	ossRegion, bizErr := detectOSSRegion(ctx, provider, bucket)
	if bizErr != nil {
		return "", bizErr
	}
	return ossRegion, nil
}

// NewOSS creates a new OSS CloudStorage instance using functional options.
// It supports both AccessKey and AssumeRole authentication.
func NewOSS(ctx context.Context, uri string, opts ...OSSOption) (CloudStorage, bizerr.BizError) {
	cfg := &OSSConfig{URI: uri}
	for _, opt := range opts {
		opt(cfg)
	}

	schema, bucket, prefix, bizError := parseS3Params(cfg.URI)
	if bizError != nil {
		return nil, bizError
	}

	var region string
	if cfg.Region != nil {
		region = *cfg.Region
	} else {
		return nil, bizerr.New(bizerr.ErrCodeParams, "OSSConfig: must provide region")
	}

	credProvider, err := buildAwsCredentialsProvider(cfg, region)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to build AWS credentials provider")
	}

	var endpoint string
	if aws.ToBool(cfg.UseInternalEndpoint) {
		endpoint = GenerateOssInternalEndpoint(region)
	} else {
		endpoint = GenerateOssEndpoint(region)
	}
	s3Client := s3.NewFromConfig(aws.Config{
		Region:       region,
		BaseEndpoint: aws.String(endpoint),
		Credentials:  credProvider,
	})

	return &S3{
		Schema: schema,
		Client: s3Client,
		Bucket: bucket,
		Prefix: prefix,
		Region: region,
	}, nil
}

// buildAliCredentialsProvider returns an aliCredetials.CredentialsProvider based on OSSConfig.
func buildAliCredentialsProvider(cfg *OSSConfig) (aliCredetials.CredentialsProvider, error) {
	if cfg.RoleArn != nil && cfg.ExternalId != nil {
		// OIDC AssumeRole mode: build OIDC config and chain role assumption
		c := new(openapicred.Config).
			SetType("oidc_role_arn").
			SetOIDCProviderArn(os.Getenv(creds.EnvOIDCProviderArn)).
			SetOIDCTokenFilePath(os.Getenv(creds.EnvOIDCTokenFile)).
			SetRoleSessionName(creds.GenerateSessionName()).
			SetRoleArn(os.Getenv(creds.EnvRoleArn))
		arnCredential, gerr := openapicred.NewCredential(c)
		return aliCredetials.CredentialsProviderFunc(func(ctx context.Context) (aliCredetials.Credentials, error) {
			if gerr != nil {
				return aliCredetials.Credentials{}, gerr
			}
			// Get base OIDC credentials
			baseRoleCred, err := arnCredential.GetCredential()
			if err != nil {
				return aliCredetials.Credentials{}, err
			}
			// Use base credentials to assume the target role
			assumeRoleCred, err := creds.AssumeRole(&creds.SessionCredentials{
				AccessKeyId:     *baseRoleCred.AccessKeyId,
				AccessKeySecret: *baseRoleCred.AccessKeySecret,
				SecurityToken:   *baseRoleCred.SecurityToken,
			}, *cfg.RoleArn, creds.GenerateSessionName(), creds.DefaultAssumeRoleDuration, cfg.ExternalId, "ap-southeast-1")
			if err != nil {
				return aliCredetials.Credentials{}, err
			}
			expirationTime, err := time.Parse(time.RFC3339, assumeRoleCred.Expiration)
			if err != nil {
				return aliCredetials.Credentials{}, err
			}
			return aliCredetials.Credentials{
				AccessKeyID:     assumeRoleCred.AccessKeyId,
				AccessKeySecret: assumeRoleCred.AccessKeySecret,
				SecurityToken:   assumeRoleCred.SecurityToken,
				Expires:         aws.Time(expirationTime),
			}, nil
		}), nil
	} else if cfg.AccessKeyID != nil && cfg.AccessKeySecret != nil {
		return aliCredetials.NewStaticCredentialsProvider(*cfg.AccessKeyID, *cfg.AccessKeySecret), nil
	}
	return nil, bizerr.New(bizerr.ErrCodeParams, "OSSConfig: must provide either (accessKeyID & accessKeySecret) or (roleArn & externalId)")
}

// detectOSSRegion gets the OSS region by calling GetBucketInfo.
func detectOSSRegion(ctx context.Context, provider aliCredetials.CredentialsProvider, bucket string) (string, bizerr.BizError) {
	ossCfg := oss.NewConfig().WithCredentialsProvider(provider).WithRegion("ap-southeast-1")
	client := oss.NewClient(ossCfg)
	request := &oss.GetBucketInfoRequest{
		Bucket: oss.Ptr(bucket),
	}
	result, err := client.GetBucketInfo(ctx, request)
	if err != nil {
		log.Error(ctx, "GetBucketInfo failed", zap.Error(err))
		var serviceErr *oss.ServiceError
		if errors.As(err, &serviceErr) && serviceErr.StatusCode == 403 {
			return "", bizerr.New(ErrCodeOssAccessDeny, errMsgOssAccessDeny)
		}
		// When calling AssumeRole with an incorrect external ID, the Alibaba Cloud SDK
		// does not return a structured error type or code that we can check.
		// Instead, it returns a generic error (e.g., errors.fundamental) with only a message.
		// Therefore, we can only detect this case by inspecting the error message text.
		if strings.Contains(err.Error(), "StatusCode: 403") {
			return "", bizerr.New(ErrCodeOssAccessDeny, errMsgOssAccessDeny)
		}
		if strings.Contains(err.Error(), "invalid field, OperationInput.Bucket") {
			return "", bizerr.New(ErrInvalidOssBucketName, errMsgInvalidOssBucketName)
		}
		return "", bizerr.WrapError(err, "GetBucketInfo failed")
	}
	location := result.BucketInfo.Location
	if location == nil {
		log.Error(ctx, "Location result is nil", zap.Any("result", result))
		return "", bizerr.WrapError(err, "GetBucketInfo failed")
	}
	region := strings.TrimPrefix(*location, "oss-")
	return region, nil
}

// buildAwsCredentialsProvider returns an aws.CredentialsProvider based on OSSConfig and region.
// It supports both AssumeRole (OIDC) and static AccessKey authentication.
func buildAwsCredentialsProvider(cfg *OSSConfig, region string) (aws.CredentialsProvider, error) {
	if cfg.RoleArn != nil && cfg.ExternalId != nil {
		return creds.NewOssOIDCAssumeCredentialsProvider(
			cfg.Duration,
			aws.ToString(cfg.RoleArn),
			cfg.ExternalId,
			region,
			cfg.RefreshWindow)
	} else if cfg.AccessKeyID != nil && cfg.AccessKeySecret != nil {
		return credentials.NewStaticCredentialsProvider(*cfg.AccessKeyID, *cfg.AccessKeySecret, ""), nil
	}
	return nil, bizerr.New(bizerr.ErrCodeParams, "OSSConfig: must provide either (accessKeyID & accessKeySecret) or (roleArn & externalId)")
}

func (s *S3) ListAll(ctx context.Context) ([]string, bizerr.BizError) {
	input := &s3.ListObjectsV2Input{
		Bucket: aws.String(s.Bucket),
	}
	if s.Prefix != nil {
		input.Prefix = s.Prefix
	}

	paginator := s3.NewListObjectsV2Paginator(s.Client, input)
	var contents []types.Object
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, bizerr.New(ErrCodeS3AccessDeny, errMsgS3AccessDeny)
		}
		contents = append(contents, page.Contents...)
	}
	var objectPaths []string
	for i := range contents {
		objectPaths = append(objectPaths, fmt.Sprintf("%s://%s/%s", s.Schema, s.Bucket, *contents[i].Key))
	}

	return objectPaths, nil
}

func (s *S3) CheckImportPermission(ctx context.Context) bizerr.BizError {
	ctx, span := Tracer.Start(ctx, "cloud_storage.s3.CheckImportPermission")
	defer span.End()
	input := &s3.ListObjectsV2Input{
		Bucket: aws.String(s.Bucket),
	}
	if s.Prefix != nil {
		input.Prefix = s.Prefix
	}

	paginator := s3.NewListObjectsV2Paginator(s.Client, input)
	for paginator.HasMorePages() {
		span.AddEvent("retrieving S3 list object page")
		// check list permission
		page, err := paginator.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return bizerr.New(ErrCodeS3AccessDeny, errMsgS3AccessDeny)
		}
		for _, content := range page.Contents {
			if isOneFile(s.Prefix) {
				// if the prefix is a file, check if the key is the same as the prefix to avoid a.csv vs a.csv.csv
				if content.Key != nil && aws.ToString(content.Key) == aws.ToString(s.Prefix) {
					if aws.ToInt64(content.Size) == 0 {
						return bizerr.New(ErrCodeImportDataFileNotFound, "Data file %s is empty", aws.ToString(s.Prefix))
					}
					span.SetAttributes(attribute.Bool("is_one_file", true))
					span.AddEvent("check get object access")
					return s.checkGetObjectPermission(ctx, content.Key, checkAccessRange)
				}
				continue
			}

			// if the prefix is a folder, select the first non-empty file
			if aws.ToInt64(content.Size) > 0 && content.Key != nil {
				span.SetAttributes(attribute.Bool("is_one_file", false))
				span.AddEvent("check get object access")
				return s.checkGetObjectPermission(ctx, content.Key, checkAccessRange)
			}
		}
	}

	if isOneFile(s.Prefix) {
		return bizerr.New(ErrCodeImportDataFileNotFound, "Can't find data file: %s", aws.ToString(s.Prefix))
	}
	return bizerr.New(ErrCodeImportEmptyFolder, errMsgImportEmptyFolder)
}

func (s *S3) checkGetObjectPermission(ctx context.Context, key *string, rangeHeader *string) bizerr.BizError {
	_, err := s.Client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(s.Bucket),
		Key:    key,
		Range:  rangeHeader,
	})
	if err != nil {
		var e *types.NoSuchKey
		if errors.As(err, &e) {
			return bizerr.New(ErrCodeImportDataFileNotFound, "Can't find data file: %s", aws.ToString(key))
		}
		log.Error(ctx, "failed to get object", zap.Error(err))
		return bizerr.New(ErrCodeS3AccessDeny, errMsgS3AccessDeny)
	}
	return nil
}

func (s *S3) RetrieveSchemaFile(ctx context.Context, schema string, table string, subDir string) (*string, bizerr.BizError) {
	schemaFilePath := getSchemaFilePath(s.Prefix, subDir, schema, table)

	input := &s3.ListObjectsV2Input{
		Bucket: aws.String(s.Bucket),
		Prefix: aws.String(schemaFilePath),
	}
	paginator := s3.NewListObjectsV2Paginator(s.Client, input)
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			log.Error(ctx, "failed to list objects from S3", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s", err.Error())
		}
		for _, obj := range page.Contents {
			if obj.Key != nil && matchSchemaFileExtension(*obj.Key, schemaFilePath) {
				result := getRelativeFilePath(s.Prefix, *obj.Key)
				return &result, nil
			}
		}
	}

	return nil, nil
}

func (s *S3) ListLimited(ctx context.Context, opts *ListFilesOptions) ([]string, bool, bizerr.BizError) {
	maxKeys := GetMaxFilesFromOptions(opts)

	resp, err := s.Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
		Bucket:  aws.String(s.Bucket),
		Prefix:  s.Prefix,
		MaxKeys: aws.Int32(int32(maxKeys)),
	})
	if err != nil {
		log.Error(ctx, "failed to list objects from S3", zap.Error(err))
		if errors.Is(err, context.DeadlineExceeded) {
			return nil, false, bizerr.New(bizerr.ErrCodeTimeout, "")
		}
		return nil, false, bizerr.New(bizerr.ErrCodeUnknown, err.Error())
	}
	var files []string
	for _, obj := range resp.Contents {
		if obj.Key != nil && !strings.HasSuffix(*obj.Key, "/") {
			files = append(files, getRelativeFilePath(s.Prefix, *obj.Key))
		}
	}
	if resp.IsTruncated != nil && *resp.IsTruncated {
		return files, true, nil
	}
	return files, false, nil
}

func parseS3Params(uri string) (schema string, bucket string, prefix *string, _ bizerr.BizError) {
	p, err := url.Parse(EscapePath(uri))
	if err != nil {
		return "", "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL")
	}
	if p.Host == "" {
		return "", "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL: Bucket name is missing or improperly specified.")
	}
	pre := strings.TrimPrefix(p.Path, "/")
	if pre != "" {
		return p.Scheme, p.Host, &pre, nil
	}
	return p.Scheme, p.Host, nil, nil
}