package storage

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

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	"github.com/aws/aws-sdk-go-v2/service/s3"

	"github.com/tidbcloud/service-proxy/pkg/checker/core"
	"github.com/tidbcloud/service-proxy/pkg/checker/util"
)

const defaultRegionHint = "us-east-1"

func init() {
	core.DefaultRegistry.Register("s3", func(req *core.CheckRequest) (core.Checker, error) {
		if req.S3 == nil {
			return nil, errors.New("s3 params required")
		}
		return NewS3Checker(req.S3), nil
	})
}

// S3Checker checks AWS S3 bucket connectivity.
type S3Checker struct {
	params *core.S3Params
}

// NewS3Checker creates a new S3 checker.
func NewS3Checker(params *core.S3Params) *S3Checker {
	return &S3Checker{params: params}
}

// Type returns the checker type.
func (c *S3Checker) Type() string {
	return "s3"
}

// Check performs S3 connectivity check.
func (c *S3Checker) Check(ctx context.Context) error {
	u, err := c.buildS3URL()
	if err != nil {
		return err
	}

	region, err := c.getBucketRegion(ctx, u.Bucket)
	if err != nil {
		return fmt.Errorf("failed to get bucket region: %w", err)
	}

	client, err := c.createClient(ctx, region)
	if err != nil {
		return fmt.Errorf("failed to create s3 client: %w", err)
	}

	if _, err = client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
		Bucket:  aws.String(u.Bucket),
		Prefix:  aws.String(u.Prefix),
		MaxKeys: aws.Int32(1),
	}); err != nil {
		return fmt.Errorf("failed to list objects in bucket %s: %w", u.Bucket, err)
	}
	return nil
}

type bucketURL struct {
	URL    string
	Prefix string
	Bucket string
}

func (c *S3Checker) buildS3URL() (*bucketURL, error) {
	s3url := strings.TrimSpace(c.params.URI)
	u, err := url.Parse(s3url)
	if err != nil {
		return nil, err
	}

	if u.Scheme != "s3" {
		return nil, errors.New("an invalid S3 endpoint URL")
	}

	parsed := util.ParseS3URL(u)
	prefix := parsed.Prefix
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	return &bucketURL{
		URL:    s3url,
		Prefix: prefix,
		Bucket: parsed.Bucket,
	}, nil
}

func (c *S3Checker) createClient(ctx context.Context, region string) (*s3.Client, error) {
	optFns := c.s3ClientOptions()
	if len(c.params.AccessKeyID) == 0 {
		return util.NewS3ClientWithAssumeRole(ctx, region, []util.AssumeRole{
			{RoleARN: c.params.RoleARN, ExternalID: c.params.ExternalID},
		}, optFns...)
	}
	return util.NewS3ClientWithStaticCredentials(
		ctx, region, c.params.AccessKeyID, c.params.SecretAccessKey, optFns...)
}

// s3ClientOptions returns the s3 client option overrides derived from the
// checker params. When a custom endpoint is configured it points the client at
// that endpoint. The addressing style depends on the provider: Alibaba Cloud
// OSS forbids path-style ("second-level domain") access and requires
// virtual-hosted style, whereas other S3-compatible stores (e.g. MinIO)
// generally need path-style. This mirrors TiCDC/BR, which disables
// force-path-style for the `alibaba` provider. It is pure and does not load any
// ambient AWS config, so it can be unit tested in isolation.
func (c *S3Checker) s3ClientOptions() []func(*s3.Options) {
	if c.params.Endpoint == "" {
		return nil
	}
	usePathStyle := !isAlibabaOSSEndpoint(c.params.Endpoint)
	return []func(*s3.Options){
		func(o *s3.Options) {
			o.BaseEndpoint = aws.String(c.params.Endpoint)
			o.UsePathStyle = usePathStyle
		},
	}
}

// aliyunOSSDomain is the domain suffix shared by all Alibaba Cloud OSS
// S3-compatible endpoints (e.g. https://oss-ap-southeast-1.aliyuncs.com). BR
// uses the same marker to detect OSS.
const aliyunOSSDomain = "aliyuncs.com"

// isAlibabaOSSEndpoint reports whether the custom endpoint points at Alibaba
// Cloud OSS, which must be accessed with virtual-hosted-style addressing.
func isAlibabaOSSEndpoint(endpoint string) bool {
	host := endpoint
	if u, err := url.Parse(endpoint); err == nil && u.Host != "" {
		host = u.Host
	}
	return strings.Contains(host, aliyunOSSDomain)
}

func (c *S3Checker) getBucketRegion(ctx context.Context, bucket string) (string, error) {
	if c.params.Region != "" {
		return c.params.Region, nil
	}
	// AWS-specific bucket-region discovery does not work for custom
	// S3-compatible endpoints (e.g. OSS); fall back to the default region
	// hint and let the configured endpoint drive the connection.
	if c.params.Endpoint != "" {
		return defaultRegionHint, nil
	}
	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(defaultRegionHint))
	if err != nil {
		return "", fmt.Errorf("failed to load AWS config: %w", err)
	}
	return manager.GetBucketRegion(ctx, s3.NewFromConfig(cfg), bucket)
}