package resource

import (
	"bytes"
	_ "embed"
	"encoding/json"
	"fmt"
	"net/url"
	"time"

	"github.com/creasty/defaults"
	"github.com/hashicorp/go-multierror"
	"github.com/jinzhu/copier"
	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
	"github.com/pkg/errors"
	"k8s.io/utils/pointer"

	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	awssts "github.com/aws/aws-sdk-go-v2/service/sts"
	apierror "github.com/pingcap/test-infra/common/errors"
	"github.com/pingcap/test-infra/common/model"
	"github.com/pingcap/test-infra/common/model/rms/request"
	"github.com/pingcap/test-infra/common/model/rms/response"
	"github.com/pingcap/test-infra/services/rms/config"
	"github.com/pingcap/test-infra/services/rms/log"
	"github.com/pingcap/test-infra/services/rms/service/provider"
	"github.com/pingcap/test-infra/services/rms/service/provider/k8s"
	"github.com/pingcap/test-infra/services/rms/service/provider/k8s/gc"
	"github.com/pingcap/test-infra/services/rms/utils"
)

type ServiceInterface interface {
	TiDBCloudServiceInterface
	CreateTestbed(ctx utils.Context, req request.CreateTestbedRequest) (*response.CreateTestbedResponse, *apierror.StatusError)
	GetTestbed(ctx utils.Context, testbed string, withDetails bool) (*response.GetTestbedResponse, *apierror.StatusError)
	CheckTestbed(ctx utils.Context, testbed string) (*response.CheckTestbedResponse, *apierror.StatusError)
	ListTestbed(ctx utils.Context) (*response.ListTestbedResponse, *apierror.StatusError)
	RestoreMonitor(ctx utils.Context, req request.RestoreMonitorRequest) (*response.RestoreMonitorResponse, *apierror.StatusError)
	AuthTestbed(ctx utils.Context, req request.AuthTestbedRequest) *apierror.StatusError
	DeleteTestbed(ctx utils.Context, testbed string) *apierror.StatusError
	GenerateCredential(ctx utils.Context, user utils.User, req request.CredentialRequest) (*response.CredentialResponse, *apierror.StatusError)
	ReclaimTestbed() *apierror.StatusError // TODO: pass in context as well
	OnCompletion(ctx utils.Context, req request.OnTestbedCompletionRequest) *apierror.StatusError
	GetLogLinks(ctx utils.Context, testbed string) (*response.GetLogLinksResponse, *apierror.StatusError)
	GetLease(ctx utils.Context, testbed string) (*response.LeaseResponse, *apierror.StatusError)
	RenewLease(ctx utils.Context, testbed string, lease request.DurationStr) (*response.RenewalLeaseResponse, *apierror.StatusError)
	GetProxy(ctx utils.Context) (*config.ProxyConfig, *apierror.StatusError)
	GetTemporaryECRImageSyncCredential(ctx utils.Context) (*response.GetTemporaryECRImageSyncCredentialResponse, *apierror.StatusError)
}

type TiDBCloudServiceInterface interface {
	GetShootKubeConfig(ctx utils.Context, req request.GetShootKubeConfigRequest) (*response.GetShootKubeConfigResponse, *apierror.StatusError)
}

var (
	_ ServiceInterface = &DefaultService{}
)

var (
	//go:embed embed/monitor-testbed.json
	monitorTestbedSpec []byte
	//go:embed embed/sdkserver.json
	sdkServerSpecTemplate []byte
)

type DefaultService struct {
	k8sProvider provider.Interface
	s3Client    *minio.Client
}

func NewDefaultService() *DefaultService {
	var s *DefaultService
	switch config.DefaultRMSConfiguration.Provider {
	case "k8s":
		k8sProvider, err := k8s.NewDefaultProvider(config.DefaultRMSConfiguration.K8s.SDKKubeConfig)
		if err != nil {
			panic(fmt.Errorf("new service error: %s", err))
		}

		minioURL, err := url.Parse(config.DefaultRMSConfiguration.MinIO.URL)
		if err != nil {
			panic(err)
		}
		s3Client, err := minio.New(minioURL.Host, &minio.Options{
			Creds:  credentials.NewStaticV4(config.DefaultRMSConfiguration.MinIO.ID, config.DefaultRMSConfiguration.MinIO.Secret, ""),
			Secure: false,
		})
		if err != nil {
			panic(err)
		}

		s = &DefaultService{k8sProvider: k8sProvider, s3Client: s3Client}
	default:
		panic(fmt.Errorf("unknown provider: %s", config.DefaultRMSConfiguration.Provider))
	}
	return s
}

func (s *DefaultService) CreateTestbed(ctx utils.Context, req request.CreateTestbedRequest) (*response.CreateTestbedResponse, *apierror.StatusError) {
	// inject sdkserver
	sdkServerItem := buildSDKServerItem(config.DefaultRMSConfiguration.RMSURL, config.DefaultRMSConfiguration.DiagServerURL, time.Now().Format(time.RFC3339))
	req.Items = append(req.Items, sdkServerItem)

	testbed, err := s.k8sProvider.CreateTestbed(ctx, req)
	if err != nil {
		return nil, err
	}
	res := &response.CreateTestbedResponse{
		Name:  testbed.Name,
		Items: []response.TestbedResponseItem{},
	}
	var errs error
	for _, item := range req.Items {
		copiedItem := model.TestBedItem{}
		_ = copier.Copy(&copiedItem, &item)
		// set the testbed name that just created
		copiedItem.Testbed = testbed.Name
		rs, err := s.k8sProvider.CreateResource(ctx, copiedItem)
		if err != nil {
			errs = multierror.Append(errs, err)
		} else {
			item := response.TestbedResponseItem{
				ResourceHandle: rs.Handle(),
			}
			res.Items = append(res.Items, item)
		}
	}
	if errs != nil {
		if err := s.k8sProvider.DeleteTestbed(ctx, testbed.Name); err != nil {
			errs = multierror.Append(errs, fmt.Errorf("delete the incompleted testbed(%s) error: %s", req.GenerateName, err))
		}
		return nil, apierror.NewInternalError(errs)
	}
	return res, nil
}

func buildSDKServerItem(rmsURL string, diagServerURL string, collectStartTime string) model.TestBedItem {
	var sdkServerItem model.TestBedItem
	sdkServerSpec := bytes.ReplaceAll(sdkServerSpecTemplate, []byte("$SDK_IMAGE"), []byte(config.DefaultRMSConfiguration.K8s.Images.SDKServer))
	sdkServerSpec = bytes.ReplaceAll(sdkServerSpec, []byte("$RMS_URL"), []byte(rmsURL))
	sdkServerSpec = bytes.ReplaceAll(sdkServerSpec, []byte("$DIAG_URL"), []byte(diagServerURL))
	sdkServerSpec = bytes.ReplaceAll(sdkServerSpec, []byte("$START_TIME"), []byte(collectStartTime))
	_ = json.Unmarshal(sdkServerSpec, &sdkServerItem)
	return sdkServerItem
}

func (s *DefaultService) GetShootKubeConfig(ctx utils.Context, req request.GetShootKubeConfigRequest) (*response.GetShootKubeConfigResponse, *apierror.StatusError) {
	return s.k8sProvider.GetShootKubeConfig(ctx, req)
}

func (s *DefaultService) GetTestbed(ctx utils.Context, testbed string, withDetails bool) (*response.GetTestbedResponse, *apierror.StatusError) {
	return s.k8sProvider.GetTestbed(ctx, testbed, withDetails)
}

func (s *DefaultService) CheckTestbed(ctx utils.Context, testbed string) (*response.CheckTestbedResponse, *apierror.StatusError) {
	return s.k8sProvider.CheckTestbed(ctx, testbed)
}

func (s *DefaultService) ListTestbed(ctx utils.Context) (*response.ListTestbedResponse, *apierror.StatusError) {
	return s.k8sProvider.ListTestbed(ctx)
}

func (s *DefaultService) RestoreMonitor(ctx utils.Context, req request.RestoreMonitorRequest) (*response.RestoreMonitorResponse, *apierror.StatusError) {
	promTarballURL, err := s.getTestbedPromTarballURL(ctx, req)
	if err != nil {
		return nil, err
	}
	testbedSpec := bytes.ReplaceAll(monitorTestbedSpec, []byte("$PROM_TARBALL_URL"), []byte(promTarballURL))
	testbedSpec = bytes.ReplaceAll(testbedSpec, []byte("$CLUSTER_TYPE"), []byte(req.ClusterType))

	createTestbedReq := request.CreateTestbedRequest{}
	if err := json.Unmarshal(testbedSpec, &createTestbedReq); err != nil {
		return nil, apierror.NewInternalError(err)
	}

	if err := defaults.Set(&createTestbedReq); err != nil {
		return nil, apierror.NewInternalError(err)
	}

	createTestbedResp, err := s.CreateTestbed(ctx, createTestbedReq)
	if err != nil {
		return nil, err
	}

	return s.k8sProvider.GetRestoredMonitorInfo(ctx, createTestbedResp.Name)
}

func (s *DefaultService) getTestbedPromTarballURL(ctx utils.Context, req request.RestoreMonitorRequest) (string, *apierror.StatusError) {
	if len(req.PrometheusTarballURL) != 0 {
		return req.PrometheusTarballURL, nil
	}

	// check file existence
	_, err := s.s3Client.StatObject(ctx.Request().Context(), gc.ArchiveBucketName, fmt.Sprintf("%s/%s/%s", gc.ArchivePathPrefix, req.Testbed, "prometheus.tar"), minio.GetObjectOptions{})
	if err != nil {
		return "", apierror.NewInternalError(errors.Wrap(err, "s3 StatObject"))
	}

	promURL, err := s.s3Client.PresignedGetObject(ctx.Request().Context(), gc.ArchiveBucketName, fmt.Sprintf("%s/%s/%s", gc.ArchivePathPrefix, req.Testbed, "prometheus.tar"), time.Hour, nil)
	if err != nil {
		return "", apierror.NewInternalError(errors.Wrap(err, "s3 PresignedGetObject"))
	}

	return promURL.String(), nil
}

func (s *DefaultService) AuthTestbed(ctx utils.Context, req request.AuthTestbedRequest) *apierror.StatusError {
	return s.k8sProvider.AuthTestbed(ctx, req)
}

// DeleteTestbed deletes the whole testbed and sub resources in it
func (s *DefaultService) DeleteTestbed(ctx utils.Context, testbed string) *apierror.StatusError {
	return s.k8sProvider.DeleteTestbed(ctx, testbed)
}

func (s *DefaultService) OnCompletion(ctx utils.Context, req request.OnTestbedCompletionRequest) *apierror.StatusError {
	return s.k8sProvider.OnCompletion(ctx, req)
}

func (s *DefaultService) GetLogLinks(ctx utils.Context, testbed string) (*response.GetLogLinksResponse, *apierror.StatusError) {
	links := make(map[string]string)
	for object := range s.s3Client.ListObjects(
		ctx.Request().Context(),
		gc.ArchiveBucketName,
		minio.ListObjectsOptions{
			Prefix:    fmt.Sprintf("%s/%s", gc.ArchivePathPrefix, testbed),
			Recursive: true,
		}) {
		u, err := s.s3Client.PresignedGetObject(
			ctx.Request().Context(),
			gc.ArchiveBucketName,
			object.Key,
			time.Hour*24*7,
			url.Values{})
		if err != nil {
			return nil, apierror.NewInternalError(err)
		}
		links[object.Key] = u.String()
	}

	return &response.GetLogLinksResponse{
		LogLinks: links,
	}, nil
}

func (s *DefaultService) GenerateCredential(ctx utils.Context, user utils.User, req request.CredentialRequest) (*response.CredentialResponse, *apierror.StatusError) {
	return s.k8sProvider.CreateCredential(ctx, user, req)
}

func (s *DefaultService) ReclaimTestbed() *apierror.StatusError {
	defer func() {
		if r := recover(); r != nil {
			log.Log.Errorf(nil, "recover from %v", r)
		}
	}()
	return s.k8sProvider.ReclaimTestbed()
}

func (s *DefaultService) GetLease(ctx utils.Context, testbed string) (*response.LeaseResponse, *apierror.StatusError) {
	resp, err := s.k8sProvider.GetLease(ctx, testbed)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (s *DefaultService) RenewLease(ctx utils.Context, testbed string, lease request.DurationStr) (*response.RenewalLeaseResponse, *apierror.StatusError) {
	resp, err := s.k8sProvider.RenewLease(ctx, testbed, lease)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (s *DefaultService) GetProxy(_ utils.Context) (*config.ProxyConfig, *apierror.StatusError) {
	return &config.DefaultRMSConfiguration.Proxy, nil
}

func (s *DefaultService) GetTemporaryECRImageSyncCredential(ctx utils.Context) (*response.GetTemporaryECRImageSyncCredentialResponse, *apierror.StatusError) {
	const region = "us-west-2"

	roleArn := config.DefaultRMSConfiguration.AwsEcrRoleArn
	if roleArn == "" {
		return nil, apierror.NewBadRequest("the credential service is not available, please make sure you are connected to the RMS address on AWS provider")
	}
	cfg, err := awsconfig.LoadDefaultConfig(ctx.Request().Context(), func(lo *awsconfig.LoadOptions) error {
		lo.Region = region
		return nil
	})
	if err != nil {
		return nil, apierror.NewInternalError(err)
	}
	client := awssts.NewFromConfig(cfg)
	resp, err := client.AssumeRole(ctx.Request().Context(), &awssts.AssumeRoleInput{
		RoleArn:         pointer.StringPtr(roleArn),
		RoleSessionName: pointer.StringPtr("rms-ecr-sync"),
	})
	if err != nil {
		return nil, apierror.NewInternalError(err)
	}
	return &response.GetTemporaryECRImageSyncCredentialResponse{
		AccessKeyId:     resp.Credentials.AccessKeyId,
		Expiration:      resp.Credentials.Expiration,
		SecretAccessKey: resp.Credentials.SecretAccessKey,
		SessionToken:    resp.Credentials.SessionToken,
	}, nil
}