package gc

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	mapset "github.com/deckarep/golang-set"
	"github.com/go-logr/logr"
	prometheusv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/resourcemanager"
	clusterctl "github.com/tidbcloud/observability/pkg/resourceworker/controllers/cluster"
	k8sapiv1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// we use this worker to destroy pvc,because prometheus operator and statefulset not support auto destroy pvc now
// it maybe removed after statefulset support destroy pvc
// ref https://github.com/kubernetes/enhancements/tree/master/keps/sig-apps/1847-autoremove-statefulset-pvcs
type GC struct {
	pollInterval time.Duration
	cli          client.Client
	log          logr.Logger
}

func NewGC(pollInterval time.Duration, log logr.Logger, cli client.Client) *GC {
	return &GC{
		pollInterval: pollInterval,
		log:          log,
		cli:          cli,
	}
}
func (g *GC) Run(ctx context.Context) {
	timer := time.NewTimer(g.pollInterval)
loop:
	for {
		select {
		case <-ctx.Done():
			break loop
		case <-timer.C:
		}
		timer.Reset(g.pollInterval)
		deletelist, pausecls, privateendpointlist, err := g.scan(ctx)
		if err != nil {
			g.log.Error(err, "error found in scan then rerun gc at next time")
			continue
		}

		err = g.delete(ctx, deletelist)
		if err != nil {
			g.log.Error(err, "error found in delete then rerun gc at next time")
			continue
		}

		err = g.delete(ctx, privateendpointlist)
		if err != nil {
			g.log.Error(err, "error found in delete private endpoint then rerun gc at next time")
			continue
		}

		err = g.pauseClusters(ctx, pausecls)
		if err != nil {
			g.log.Error(err, "error found in pause clusters then rerun gc at next time")
			continue
		}

		g.log.Info("gc worker done")
	}
}

func (g *GC) getPromeSet(ctx context.Context) (mapset.Set, error) {
	var plist prometheusv1.PrometheusList
	err := g.cli.List(ctx, &plist)
	if err != nil {
		return nil, err
	}

	promeset := mapset.NewSet()
	for _, p := range plist.Items {
		promeset.Add(p.Name)
	}

	return promeset, nil
}

func (g *GC) scanPVC(ctx context.Context, log logr.Logger) ([]client.Object, error) {
	// list all prometheus crd and pick prometheus name into list
	log = log.WithName("PVC")

	promeset, err := g.getPromeSet(ctx)
	if err != nil {
		return nil, err
	}
	log.V(1).Info("prometheus ", "set", promeset)

	// list all pvc and pick also
	var pvclist k8sapiv1.PersistentVolumeClaimList
	err = g.cli.List(ctx, &pvclist, &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{"app.kubernetes.io/name": "prometheus"}),
	})
	if err != nil {
		return nil, err
	}
	pvcpromeset := mapset.NewSet()
	for _, p := range pvclist.Items {
		if lb, ok := p.Labels["operator.prometheus.io/name"]; ok {
			pvcpromeset.Add(lb)
		}
	}
	log.V(1).Info("pvc prometheus ", "set", pvcpromeset)

	var deletelist []client.Object
	deleteset := pvcpromeset.Difference(promeset)
	for i := range pvclist.Items {
		if lb, ok := pvclist.Items[i].Labels["operator.prometheus.io/name"]; ok {
			if deleteset.Contains(lb) {
				deletelist = append(deletelist, &pvclist.Items[i])
			}
		}
	}
	log.V(1).Info("gc will delete this pvc", "len", len(deletelist), "set", deletelist)
	return deletelist, nil

}

type clusterCanBeProcessedChecker func(cluster observabilitypingcapcomv1.Cluster) bool

func checkIfClusterCanBeDeleted(cluster observabilitypingcapcomv1.Cluster) bool {
	if cluster.Status.Phase == constants.TO_BE_DELETED {
		tstr := cluster.Labels[constants.LabelTimestampToDelete]
		t, _ := strconv.ParseInt(tstr, 10, 64)

		// do not delete cluster unless last snapshot has been upload to s3
		snapshot := cluster.Annotations[constants.AnnotationNewDataSnapshot]
		if t < time.Now().Unix() && len(snapshot) == 0 {
			return true
		}
	}

	return false
}

func checkIfClusterCanBePaused(cluster observabilitypingcapcomv1.Cluster) bool {
	if clusterctl.IsSharedMode(&cluster) {
		return false
	}

	if cluster.Status.Phase == constants.TO_BE_PAUSED {
		tstr := cluster.Labels[constants.LabelTimestampToPause]
		t, _ := strconv.ParseInt(tstr, 10, 64)
		snapshot := cluster.Annotations[constants.AnnotationNewDataSnapshot]
		if t < time.Now().Unix() && len(snapshot) == 0 {
			return true
		}
	}

	return false
}

func (g *GC) scanCluster(ctx context.Context, log logr.Logger, labels client.HasLabels, checker clusterCanBeProcessedChecker) ([]client.Object, error) {
	log = log.WithName("Cluster")

	if len(labels) == 0 || checker == nil {
		return nil, errors.New("param labels or checker is invalid")
	}

	var clusterlist observabilitypingcapcomv1.ClusterList
	err := g.cli.List(ctx, &clusterlist, &labels)
	if err != nil {
		return nil, err
	}
	var list []client.Object
	for i, c := range clusterlist.Items {
		if checker(c) {
			list = append(list, &clusterlist.Items[i])
			continue
		}
	}
	log.V(1).Info("gc will process clusters", "len", len(list), "set", list)
	return list, nil
}

func (g *GC) scanPrivateEndpoint(ctx context.Context, log logr.Logger) ([]client.Object, error) {
	log = log.WithName("PrivateEndpoint")

	var endpointlist observabilitypingcapcomv1.PrivateEndpointList
	err := g.cli.List(ctx, &endpointlist)
	if err != nil {
		return nil, err
	}
	var list []client.Object
	for i, c := range endpointlist.Items {
		endpointlabels := c.GetLabels()
		project := endpointlabels[constants.LabelProjectID]
		tenant := endpointlabels[constants.LabelTenantID]
		biztype := endpointlabels[constants.LabelBizType]
		var clusterList observabilitypingcapcomv1.ClusterList
		listOptions := &client.ListOptions{
			Namespace:     fmt.Sprintf("%s-%s", biztype, tenant),
			LabelSelector: labels.SelectorFromSet(map[string]string{constants.LabelProjectID: project}),
		}

		err := g.cli.List(ctx, &clusterList, listOptions)
		if err != nil {
			return nil, err
		}
		if len(clusterList.Items) == 0 {
			list = append(list, &endpointlist.Items[i])
		}

	}
	log.V(1).Info("gc will process private endpoint", "len", len(list), "set", list)
	return list, nil
}

func (g *GC) scan(ctx context.Context) ([]client.Object, []client.Object, []client.Object, error) {
	log := g.log.WithName("scan")
	var deletelist []client.Object
	pvcs, err := g.scanPVC(ctx, log)
	if err != nil {
		return nil, nil, nil, err
	}
	deletelist = append(deletelist, pvcs...)

	labels := client.HasLabels{constants.LabelTimestampToDelete}
	deletecls, err := g.scanCluster(ctx, log, labels, checkIfClusterCanBeDeleted)
	if err != nil {
		return nil, nil, nil, err
	}
	deletelist = append(deletelist, deletecls...)

	labels = client.HasLabels{constants.LabelTimestampToPause}
	pausecls, err := g.scanCluster(ctx, log, labels, checkIfClusterCanBePaused)
	if err != nil {
		return nil, nil, nil, err
	}

	privateEndpoints, err := g.scanPrivateEndpoint(ctx, log)
	if err != nil {
		return nil, nil, nil, err
	}

	//return the obj list
	return deletelist, pausecls, privateEndpoints, nil
}
func (g *GC) delete(ctx context.Context, list []client.Object) error {
	log := g.log.WithName("delete")
	var err error
	for _, item := range list {
		err = g.cli.Delete(ctx, item)
		if err != nil {
			log.V(1).Error(err, "delete item failed", "name", item.GetName(), "namespace", item.GetNamespace())
			return err
		}
	}
	return nil

}

func (g *GC) pauseClusters(ctx context.Context, list []client.Object) error {
	log := g.log.WithName("pauseClusters")
	var err error
	for _, item := range list {
		cluster, ok := item.(*observabilitypingcapcomv1.Cluster)
		if !ok || cluster == nil {
			continue
		}

		if clusterctl.IsSharedMode(cluster) {
			continue
		}

		cluster.Spec.Prometheus.Replicas = 0
		clusterRW := resourcemanager.GetClusterReaderWriter(ctx)
		err = clusterRW.PauseCluster(ctx, cluster)
		if err != nil {
			log.V(1).Error(err, "pause cluster failed", "name", cluster.GetName(), "namespace", cluster.GetNamespace())
			return err
		}
	}
	return nil
}