package cluster

import (
	"context"
	"io"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

const (
	// The key of CA certificate in bucket
	caCertificateS3Key string = "ca.pem"

	// Set response-content-disposition to 'attachment' to make browser don't attempt to render file
	respContentDispositionAttachment string = "attachment"

	// The default duration time to expires presigned URL
	defaultCACertPresignExpires = 1 * time.Minute
)

func (s *serviceImpl) clusterCertsBucket() (bucket string, err bizerr.BizError) {
	bucket = s.cfg.BizConfig.ClusterCertsConfig.Bucket
	if bucket == "" {
		return "", bizerr.New(bizerr.GeneralInternalErrorCode, "cluster certs S3 bucket is not configured")
	}
	return bucket, nil
}

func (s *serviceImpl) GetClusterCACertificate(
	ctx context.Context,
	clusterId uint64,
) ([]byte, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.GetClusterCACertificate")
	defer span.End()

	if clusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	_, err := s.clusterRepo.Get(ctx, clusterId, false)
	if err != nil {
		return nil, err
	}

	bucket, err := s.clusterCertsBucket()
	if err != nil {
		return nil, err
	}

	out, err := s.s3GetClient.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(caCertificateS3Key),
	})
	if err != nil {
		return nil, err
	}
	defer out.Body.Close()

	certBytes, ioErr := io.ReadAll(out.Body)
	if ioErr != nil {
		return nil, bizerr.New(bizerr.GeneralInternalErrorCode, "failed to read CA certificate from S3: %v", ioErr)
	}

	return certBytes, nil
}

func (s *serviceImpl) GetClusterCACertificateDownloadURL(
	ctx context.Context,
	clusterId uint64,
) (string, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.GetClusterCACertificateDownloadURL")
	defer span.End()

	if clusterId == 0 {
		return "", bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	_, err := s.clusterRepo.Get(ctx, clusterId, false)
	if err != nil {
		return "", err
	}

	bucket, err := s.clusterCertsBucket()
	if err != nil {
		return "", err
	}

	preSignResp, err := s.s3PresignClient.PresignGetObject(ctx, &s3.GetObjectInput{
		Bucket:                     aws.String(bucket),
		Key:                        aws.String(caCertificateS3Key),
		ResponseContentDisposition: aws.String(respContentDispositionAttachment),
	}, s3.WithPresignExpires(defaultCACertPresignExpires))
	if err != nil {
		return "", err
	}

	return preSignResp.URL, nil
}

func (s *serviceImpl) SetClientCaCertificate(
	ctx context.Context,
	req *globaldedicatedv1.SetClientCaCertificateRequest,
) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "Domain.ResumeCluster")
	defer span.End()

	if req.CaCertificate == "" {
		return bizerr.New(bizerr.ErrCodeParams, "ca certificate is required")
	}

	if req.ClusterId == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	cluster, err := s.clusterRepo.Get(ctx, req.ClusterId, false)
	if err != nil {
		return err
	}

	region, err := s.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return err
	}

	if region.State != globalcommonv1.Region_ACTIVE {
		return bizerr.New(bizerr.ErrCodePermission, "region is not active")
	}

	_, err = s.regionServerCli.SetClientCaCertificate(ctx, region, &regionaldedicatedv1.SetClientCaCertificateRequest{
		ClusterId: strconv.FormatUint(req.ClusterId, 10),
		Ca:        req.CaCertificate,
	})
	if err != nil {
		return err
	}

	return nil
}

func (s *serviceImpl) GetClientCaCertificate(
	ctx context.Context,
	req *globaldedicatedv1.GetClientCaCertificateRequest,
) (*globaldedicatedv1.GetClientCaCertificateResponse, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Domain.ResumeCluster")
	defer span.End()

	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	cluster, err := s.clusterRepo.Get(ctx, req.ClusterId, false)
	if err != nil {
		return nil, err
	}

	region, err := s.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	if region.State != globalcommonv1.Region_ACTIVE {
		return nil, bizerr.New(bizerr.ErrCodePermission, "region is not active")
	}

	ca, err := s.regionServerCli.GetCaCertificate(ctx, region, &regionaldedicatedv1.GetCaCertificateRequest{
		ClusterId: strconv.FormatUint(req.ClusterId, 10),
		Usage:     regionaldedicatedv1.CaCertificate_CLIENT,
	})
	if err != nil {
		return nil, err
	}
	return &globaldedicatedv1.GetClientCaCertificateResponse{CaCertificate: ca.Ca}, nil
}