package networkpolicy

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	apitypes "k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/util/workqueue"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/infra-provider/network-policy-controller/pkg/types"
)

type S3CIDRPoller struct {
	logger logr.Logger

	kc client.Client

	region string
	c      ec2.DescribePrefixListsAPIClient

	cidrs []string

	lock   sync.RWMutex
	synced chan struct{}
}

func NewS3CIDRPoller(logger logr.Logger, kc client.Client, c ec2.DescribePrefixListsAPIClient, region string) *S3CIDRPoller {
	return &S3CIDRPoller{
		kc:     kc,
		c:      c,
		logger: logger,
		region: region,
		synced: make(chan struct{}),
	}
}

func (p *S3CIDRPoller) Poll(ctx context.Context) (bool, error) {
	name := fmt.Sprintf("com.amazonaws.%s.s3", p.region)
	lists, err := p.c.DescribePrefixLists(ctx, &ec2.DescribePrefixListsInput{
		Filters: []ec2types.Filter{
			{
				Name: aws.String("prefix-list-name"),
				Values: []string{
					name,
				},
			},
		},
	})
	if err != nil {
		return false, err
	}
	if len(lists.PrefixLists) != 1 {
		return false, fmt.Errorf("expected length of prefix list is 1, current is %v", len(lists.PrefixLists))
	}
	var cidrs []string
	for _, item := range lists.PrefixLists {
		cidrs = append(cidrs, item.Cidrs...)
	}

	sort.Strings(cidrs)
	if !reflect.DeepEqual(p.cidrs, cidrs) {
		p.lock.Lock()
		p.cidrs = cidrs
		p.lock.Unlock()
		return true, nil
	}
	return false, nil
}

func (p *S3CIDRPoller) CIDRs() []string {
	p.lock.RLock()
	defer p.lock.RUnlock()
	return p.cidrs
}

func (p *S3CIDRPoller) Start(ctx context.Context, h handler.EventHandler, q workqueue.RateLimitingInterface, ps ...predicate.Predicate) error {
	go func() {
		if err := wait.PollImmediateUntilWithContext(ctx, 10*time.Second, func(ctx context.Context) (bool, error) {
			if _, err := p.Poll(ctx); err != nil {
				p.logger.Error(err, "sync cidr info failed, retry")
				return false, nil
			}
			return true, nil
		}); err != nil {
			p.logger.Error(err, "failed to sync cidr info")
			return
		}

		close(p.synced)

		if err := wait.PollInfiniteWithContext(ctx, 1*time.Minute, func(ctx context.Context) (bool, error) {
			changed, err := p.Poll(ctx)
			if err != nil {
				p.logger.Error(err, "poll cidr info failed")
				return false, nil
			}
			if changed {
				nss := corev1.NamespaceList{}
				if err := p.kc.List(ctx, &nss, &client.MatchingLabels{
					types.LabelKeyNamespaceType: types.LabelValNamespaceTypeTiDB,
				}); err != nil {
					p.logger.Error(err, "list namespaces failed")
					return false, nil
				}
				// skip event handler and add req into queue directly
				for _, ns := range nss.Items {
					for _, policyName := range AllowToS3PolicyNames {
						q.Add(ctrl.Request{
							NamespacedName: apitypes.NamespacedName{
								Namespace: ns.Name,
								Name:      policyName,
							},
						})
					}
				}
			}
			return false, nil
		}); err != nil {
			if errors.Is(err, wait.ErrWaitTimeout) {
				p.logger.Info("poller is stopped as expected")
			} else {
				p.logger.Error(err, "poller is stopped unexpectedly")
			}
		}
	}()

	return nil
}

func (p *S3CIDRPoller) WaitForSync(ctx context.Context) error {
	select {
	case <-p.synced:
		return nil
	case <-ctx.Done():
		if errors.Is(ctx.Err(), context.Canceled) {
			return errors.New("cidr sync is canceled")
		}
		return errors.New("timed out waiting for cidr to be synced")
	}
}