package eks

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"k8s.io/utils/pointer"

	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkservicemigration/task"
)

type RejectEndpoints struct {
	*task.BaseTask
	awsCli aws.Client
}

func NewRejectEndpoints(c *task.Context, awsCli aws.Client) *RejectEndpoints {
	return &RejectEndpoints{BaseTask: &task.BaseTask{Context: c}, awsCli: awsCli}
}

func (t *RejectEndpoints) Name() string {
	return "RejectEndpoints"
}

func (t *RejectEndpoints) Create(ctx context.Context) deploy.Result {
	if err := t.rejectAllVPCEndpoints(ctx, t.PrivateLinkService.Status.EndpointService.ServiceID); err != nil {
		return deploy.Panic(err).WithRequeuAfter(time.Second * 10)
	}
	return deploy.Done()
}

func (t *RejectEndpoints) rejectAllVPCEndpoints(ctx context.Context, svcID string) error {
	conns, err := t.getAllVPCEndpointConns(ctx, svcID)
	if err != nil {
		return err
	}
	rejectList := []string{}
	for _, conn := range conns {
		t.Logger.Info("checking vpc endpoint", "conn", conn)
		if isUnrejected(&conn) {
			rejectList = append(rejectList, *conn.VpcEndpointId)
		}
	}
	t.Logger.Info("rejecting vpc endpoints", "rejectList", rejectList)
	if len(rejectList) == 0 {
		return nil
	}
	if err := t.rejectVPCEndpoints(ctx, svcID, rejectList...); err != nil {
		return err
	}

	return nil
}

func (t *RejectEndpoints) rejectVPCEndpoints(ctx context.Context, svcID string, rejectList ...string) error {
	cli := t.awsCli.EC2(&aws.AssumedRole{RoleARN: t.RoleARN()})
	res, err := cli.RejectVpcEndpointConnections(ctx,
		&ec2.RejectVpcEndpointConnectionsInput{
			ServiceId:      &svcID,
			VpcEndpointIds: rejectList,
		}, func(opts *ec2.Options) {
			opts.Region = t.PrivateLinkService.Spec.Region
		})
	if err != nil {
		return err
	}

	if len(res.Unsuccessful) != 0 {
		return fmt.Errorf("reject some vpc endpoints(%v) failed", res.Unsuccessful)
	}

	return nil
}

func (t *RejectEndpoints) getAllVPCEndpointConns(ctx context.Context, svcID string) ([]ec2types.VpcEndpointConnection, error) {
	cli := t.awsCli.EC2(&aws.AssumedRole{RoleARN: t.RoleARN()})
	res, err := cli.DescribeVpcEndpointConnections(ctx,
		&ec2.DescribeVpcEndpointConnectionsInput{
			Filters: []ec2types.Filter{
				{
					Name: pointer.String("service-id"),
					Values: []string{
						svcID,
					},
				},
			},
		}, func(opts *ec2.Options) {
			opts.Region = t.PrivateLinkService.Spec.Region
		})
	if err != nil {
		return nil, err
	}

	return res.VpcEndpointConnections, nil
}

const (
	epStatePendingAcceptance = "pendingacceptance"
	epStatePending           = "pending"
	epStateAvailable         = "available"
	epStateDeleting          = "deleting"
	epStateDeleted           = "deleted"
	epStateRejected          = "rejected"
	epStateFailed            = "failed"
	epStateExpired           = "expired"
)

func isUnrejected(conn *ec2types.VpcEndpointConnection) bool {
	state := strings.ToLower(string(conn.VpcEndpointState))
	switch state {
	case epStateAvailable, epStatePending, epStatePendingAcceptance:
		return true
	case epStateDeleting, epStateDeleted, epStateRejected, epStateFailed, epStateExpired:
	}

	return false
}