package aws

import (
	"context"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/eks"
	"github.com/pingcap/errors"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
)

type AWSEKSClientProxy interface {
	GetClient(ctx context.Context, req *GetAWSClientReq) (*eks.EKS, error)
}

type awsEKSClientProxyImpl struct {
	lock sync.Mutex

	// k : accountID-region, v : eks client
	clientM map[string]*eks.EKS
}

func NewAWSEKSClientProxy() AWSEKSClientProxy {
	return &awsEKSClientProxyImpl{clientM: map[string]*eks.EKS{}}
}

func (b *awsEKSClientProxyImpl) GetClient(ctx context.Context, req *GetAWSClientReq) (*eks.EKS, error) {
	b.lock.Lock()
	defer b.lock.Unlock()

	key := fmt.Sprintf("%s-%s", req.AccountID, req.Region)
	if client, ok := b.clientM[key]; ok {
		return client, nil
	}
	awsSessionOpts := session.Options{
		Config: *aws.NewConfig().WithCredentialsChainVerboseErrors(true).WithRegion(req.Region),
	}
	ses, err := session.NewSessionWithOptions(awsSessionOpts)
	if err != nil {
		return nil, errors.AddStack(err)
	}
	roleARN := req.GenerateDataflowRoleArn()
	// always use the dfs role which has the all permissions of all eks
	creds := stscreds.NewCredentials(ses, roleARN, func(p *stscreds.AssumeRoleProvider) {
		p.RoleARN = roleARN
	})
	client := eks.New(ses, &aws.Config{
		Credentials: creds,
		HTTPClient: &http.Client{
			Timeout: time.Duration(service_config.Global().AwsConfig.S3SDKRequestTimeout) * time.Second,
		},
	})
	b.clientM[key] = client
	return client, nil
}