package cluster

import (
	"context"

	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"

	vmv1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
	"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/resourceworker/controllers/reconciler"
	k8sapiv1 "k8s.io/api/core/v1"
	k8snetworkv1 "k8s.io/api/networking/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

func (cls *Cluster) MakeMonitorResource(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	if IsSharedMode(clusterv1) {
		if !RejectInsertIngress(clusterv1) {
			var externalWriteService k8sapiv1.Service
			if err := cls.getAndCreate(ctx, clusterv1.Namespace, vmInsertSvcPrefix+clusterv1.Name, &externalWriteService, clusterv1, makeVMInsertExternalService); err != nil {
				return err
			}
		}

		if !RejectReadIngress(cls, clusterv1) {
			var externalReadService k8sapiv1.Service
			if err := cls.getAndCreate(ctx, clusterv1.Namespace, vmSelectSvcPrefix+clusterv1.Name, &externalReadService, clusterv1, makeVMSelectExternalService); err != nil {
				return err
			}
		}

		needVM, err := VMAlertNeeded(ctx, cls, clusterv1)
		if err != nil {
			return err
		}
		if needVM {
			var vmAlert vmv1.VMAlert
			if err := cls.getAndCreate(ctx, clusterv1.Namespace, clusterv1.Name, &vmAlert, clusterv1, makeVMAlert); err != nil {
				return err
			}
		}
	} else {
		//create prometheus crd
		var prome prometheusv1.Prometheus
		if err := cls.getAndCreate(ctx, clusterv1.Namespace, clusterv1.Spec.Prometheus.Name, &prome, clusterv1, makePrometheus); err != nil {
			return err
		}

		//create thanos secret if export metrics to object storage
		if clusterv1.Spec.Prometheus.MetricsToObjectStorage {
			var oldsecret k8sapiv1.Secret
			if err := cls.getAndCreate(ctx, clusterv1.Namespace, clusterv1.Spec.Prometheus.Name, &oldsecret, clusterv1, makeThanosObjectConfigSecret); err != nil {
				return err
			}
		}

		if len(clusterv1.Spec.Endpoint.Name) != 0 {
			//create service crd
			var prometheusService k8sapiv1.Service
			if err := cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name, &prometheusService, clusterv1, makePrometheusService); err != nil {
				return err
			}
		}
	}

	if clusterv1.Spec.Endpoint.Shared {
		// create read ingress
		if err := cls.makeIngress(ctx, clusterv1); err != nil {
			return err
		}
	}

	return nil
}

func (cls *Cluster) makeIngress(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	var ingress k8snetworkv1.Ingress

	if IsSharedMode(clusterv1) {
		return errors.OfNullable(nil).BreakIf(func() error {
			if !RejectReadIngress(cls, clusterv1) {
				return cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+readSuffix, &ingress, clusterv1, makeVMSelectIngress)
			}
			return nil
		}).BreakIf(func() error {
			if !RejectReadIngress(cls, clusterv1) {
				return cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+openReadSuffix, &ingress, clusterv1, makeVMOpenSelectIngress)
			}
			return nil
		}).BreakIf(func() error {
			if !RejectInsertIngress(clusterv1) {
				return cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name, &ingress, clusterv1, makeVMInsertIngress)
			}
			return nil
		}).BreakIf(func() error {
			if SupportServerless(clusterv1) {
				return cls.getAndCreate(ctx, clusterv1.Namespace, serverlessPrefix+clusterv1.Spec.Endpoint.Name, &ingress, clusterv1, makeServerLessInsertIngress)
			}
			return nil
		}).Present()
	} else {
		return errors.OfNullable(nil).BreakIf(func() error {
			return cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+readSuffix, &ingress, clusterv1, makePrometheusIngressRead)
		}).BreakIf(func() error {
			return cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+openReadSuffix, &ingress, clusterv1, makePrometheusIngressOpenRead)
		}).BreakIf(func() error {
			return cls.getAndCreate(ctx, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name, &ingress, clusterv1, makePrometheusIngressWrite)
		}).Present()
	}
}

func (cls *Cluster) DeleteMonitorResource(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	log := log.FromContext(ctx).WithName("DeleteMonitorResource")

	if clusterv1.Spec.Endpoint.Shared {
		// create read ingress
		if err := cls.deleteIngress(ctx, log, clusterv1); err != nil {
			return err
		}
	}

	if IsSharedMode(clusterv1) {
		var vmAlert vmv1.VMAlert
		if err := cls.getAndDelete(ctx, log, clusterv1.Namespace, clusterv1.Name, &vmAlert); err != nil {
			return err
		}

		var externalWriteService k8sapiv1.Service
		if err := cls.getAndDelete(ctx, log, clusterv1.Namespace, vmInsertSvcPrefix+clusterv1.Name, &externalWriteService); err != nil {
			return err
		}
		var externalReadService k8sapiv1.Service
		if err := cls.getAndDelete(ctx, log, clusterv1.Namespace, vmSelectSvcPrefix+clusterv1.Name, &externalReadService); err != nil {
			return err
		}
		return nil
	} else {
		//delete prometheus service crd
		var promesvc k8sapiv1.Service
		if err := cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name, &promesvc); err != nil {
			return err
		}

		//delete prometheus crd
		var prome prometheusv1.Prometheus
		if err := cls.getAndDelete(ctx, log, clusterv1.Namespace, clusterv1.Spec.Prometheus.Name, &prome); err != nil {
			return err
		}
		//delete thanos secret
		var thanossecret k8sapiv1.Secret
		if err := cls.getAndDelete(ctx, log, clusterv1.Namespace, clusterv1.Spec.Prometheus.Name, &thanossecret); err != nil {
			return err
		}
		return nil
	}

}

func (cls *Cluster) deleteIngress(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	var ingress k8snetworkv1.Ingress

	if IsSharedMode(clusterv1) {
		return errors.OfNullable(nil).BreakIf(func() error {
			if !RejectReadIngress(cls, clusterv1) {
				return cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+readSuffix, &ingress)
			}
			return nil
		}).BreakIf(func() error {
			if !RejectReadIngress(cls, clusterv1) {
				return cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+openReadSuffix, &ingress)
			}
			return nil
		}).BreakIf(func() error {
			if !RejectInsertIngress(clusterv1) {
				return cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name, &ingress)
			}
			return nil
		}).BreakIf(func() error {
			if SupportServerless(clusterv1) {
				return cls.getAndDelete(ctx, log, clusterv1.Namespace, serverlessPrefix+clusterv1.Spec.Endpoint.Name, &ingress)
			}
			return nil
		}).Present()
	} else {
		return errors.OfNullable(nil).BreakIf(func() error {
			return cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+readSuffix, &ingress)
		}).BreakIf(func() error {
			return cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name+openReadSuffix, &ingress)
		}).BreakIf(func() error {
			return cls.getAndDelete(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name, &ingress)
		}).Present()
	}

}

func (cls *Cluster) GetMonitorCrd(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	if IsSharedMode(clusterv1) {
		return getVMCrd(ctx, cls, clusterv1)
	} else {
		return getPrometheusCrd(ctx, cls, clusterv1)
	}
}

func (cls *Cluster) UpdateMonitorCRD(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) error {
	if IsSharedMode(clusterv1) {
		return cls.updateVM(ctx, clusterv1, obj)
	} else {
		return cls.updatePrometheus(ctx, obj, clusterv1)
	}
}

func (cls *Cluster) UpdateMonitorOnPauseStatus(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object, paused bool) error {
	if IsSharedMode(clusterv1) {
		return cls.updateVMOnPauseStatus(ctx, obj, clusterv1, paused)
	}
	return cls.updatePrometheusReplicas(ctx, obj, clusterv1)
}

func (cls *Cluster) ConvertMonitorStatus(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object, checkTimeout bool) (bool, bool, error) {
	var clusterTime string
	if checkTimeout {
		clusterTime = clusterv1.Status.UpdateAt
	}

	if IsSharedMode(clusterv1) {
		return true, false, nil
	} else {
		return prometheusStatus(ctx, clusterv1, clusterTime, obj)
	}
}

func (cls *Cluster) CheckIfMonitorPaused(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (bool, error) {
	if IsSharedMode(clusterv1) {
		return cls.checkIfVMPaused(ctx, clusterv1, obj)
	}

	prometheus, ok := obj.(*prometheusv1.Prometheus)
	if !ok || prometheus == nil {
		return false, errors.New(errors.ParameterInvalid, "param obj invalid")
	}
	return checkIfPrometheusPausedByReplicas(ctx, prometheus), nil
}

func (cls *Cluster) CheckIfMonitorReadyByReplicas(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (bool, error) {
	if IsSharedMode(clusterv1) {
		return cls.checkIfVMReady(ctx, clusterv1, obj)
	}

	prometheus, ok := obj.(*prometheusv1.Prometheus)
	if !ok || prometheus == nil {
		return false, errors.New(errors.ParameterInvalid, "param obj invalid")
	}
	return checkIfPrometheusReadyByReplicas(ctx, prometheus)
}

func (cls *Cluster) ObserveMonitorResource(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) (reconciler.SubResourceObservation, client.Object, error) {
	if IsSharedMode(clusterv1) {
		return observeVMResource(ctx, cls, clusterv1)
	} else {
		return observePrometheusResource(ctx, cls, clusterv1)
	}
}

func (cls *Cluster) GetMonitorAZ(clusterv1 *observabilitypingcapcomv1.Cluster) string {
	if IsSharedMode(clusterv1) {
		return getVMAZ(clusterv1)
	} else {
		return getPrometheusAZ(clusterv1)
	}
}

func (cls *Cluster) MonitorEqual(ctx context.Context, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster) bool {
	if IsSharedMode(clusterv1) {
		return equalVM(ctx, cls, obj, clusterv1)
	} else {
		return equalPrometheus(ctx, cls, obj, clusterv1)
	}
}

func (cls *Cluster) RestartWithSnapshot(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	if IsSharedMode(clusterv1) {
		return nil
	} else {
		return cls.updatePrometheusTSDB(ctx, clusterv1)
	}
}

func (cls *Cluster) updateVM(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) error {
	needVM, err := VMAlertNeeded(ctx, cls, clusterv1)
	if err != nil {
		return err
	}
	if !needVM || obj == nil {
		return nil
	}
	_, err = controllerutil.CreateOrUpdate(ctx, cls.cli, obj, func() error {
		expected, err := makeVMAlert(cls, clusterv1)
		if err != nil {
			return err
		}
		vmalert := obj.(*vmv1.VMAlert)
		expectedSpec := expected.(*vmv1.VMAlert).Spec
		vmalert.Spec = expectedSpec
		return nil
	})
	return err
}

func (cls *Cluster) checkIfVMPaused(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (bool, error) {
	vmrule := &vmv1.VMRule{}
	rPaused := false
	err := cls.cli.Get(ctx, client.ObjectKey{Namespace: clusterv1.Namespace, Name: clusterv1.Name}, vmrule)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			return false, err
		}
		rPaused = true // consider vmrule paused if not found
	} else {
		rPaused = vmrule.Labels[constants.LabelClusterPaused] == "true"
	}
	vmalert, ok := obj.(*vmv1.VMAlert)
	if !ok || vmalert == nil {
		return false, errors.New(errors.ParameterInvalid, "param obj invalid")
	}

	// vm is paused when
	// 1. vmrule is paused
	// 2. vmalert is not needed or vmalert's replicaCount is 0
	return rPaused &&
		(!VMAlertNeededByVMRule(clusterv1, vmrule) || (vmalert.Spec.ReplicaCount != nil && *vmalert.Spec.ReplicaCount == 0)), nil
}

func (cls *Cluster) checkIfVMReady(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (bool, error) {
	vmrule := &vmv1.VMRule{}
	rReady := false
	err := cls.cli.Get(ctx, client.ObjectKey{Namespace: clusterv1.Namespace, Name: clusterv1.Name}, vmrule)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			return false, err
		}
		rReady = true // consider vmrule ready if not found
	} else {
		rReady = vmrule.Labels[constants.LabelClusterPaused] == ""
	}
	vmalert, ok := obj.(*vmv1.VMAlert)
	if !ok || vmalert == nil {
		return false, errors.New(errors.ParameterInvalid, "param obj invalid")
	}

	// vm is ready when
	// 1. vmrule is ready
	// 2. vmalert is not needed or vmalert's replicaCount is nil or not 0
	return rReady &&
		(!VMAlertNeededByVMRule(clusterv1, vmrule) || (vmalert.Spec.ReplicaCount == nil || *vmalert.Spec.ReplicaCount != 0)), nil
}